python►notes

python argparse

创建解析对象

1
parser = argparse.ArgumentParser()

kv

1
2
parser.add_argument('-a', '--age', required=True)      # 必选
parser.add_argument('-p', '--product', required=False)  # 非必选

example

1
2
3
4
5
6
7
8
9
10
11
12
13
>>> import argparse
>>> parser = argparse.ArgumentParser()
>>> parser.add_argument('-a', '--age', required=True)
...
>>> parser.add_argument('-p', '--product', required=False)
...
>>> parser.parse_args('-a 10'.split())
Namespace(age='10', product=None)
>>> parser.parse_args('-a 10 -p wii'.split())
Namespace(age='10', product='wii')
>>> parser.parse_args('-p wii'.split())
usage: [-h] -a AGE [-p PRODUCT]
: error: the following arguments are required: -a/--age

标记

1
2
3
parser.add_argument('-a', '--age', action='store_true')         # 如果指定则为 True, 否则为 False
parser.add_argument('-n', '--name', action='store_false')       # 如果指定则为 False, 否则为 True
parser.add_argument('-m', '--price', action='store_const', const=10)  # 如果指定则为 const 的值, 否则为 None

example

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
>>> import argparse
>>> parser = argparse.ArgumentParser()
>>> parser = argparse.ArgumentParser()
>>> parser.add_argument('-a', '--age', action='store_true')
...
>>> parser.add_argument('-n', '--name', action='store_false')
...
>>> parser.add_argument('-m', '--price', action='store_const', const=10)
...
>>> parser.parse_args('-anm'.split())
Namespace(age=True, name=False, price=10)
>>> parser.parse_args('-a'.split())
Namespace(age=True, name=True, price=None)
>>> parser.parse_args('-m'.split())
Namespace(age=False, name=True, price=10)
>>> parser.parse_args('-n'.split())
Namespace(age=False, name=False, price=None)
>>> parser.parse_args(''.split())
Namespace(age=False, name=True, price=None)
linux►distro►ubuntu

ubuntu - vnc

安装 Desktop

1
2
3
4
5
6
sudo apt update && sudo apt upgrade 
sudo apt install tasksel -y 
sudo tasksel	# 选择安装 ubuntu-desktop
sudo systemctl set-default graphical.target 
# 重启电脑
sudo reboot

安装 TigerVNCServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
sudo apt install tigervnc-standalone-server tigervnc-common

# 设置 vnc 密码
vncpasswd

# 设置 vnc server
sudo vi  ~/.vnc/xstartup
# 内容如下
#!/bin/sh
xrdb $HOME/.Xresources
vncconfig -iconic &
dbus-launch --exit-with-session gnome-session &

# 启动 vnc server
vncserver -localhost no

操作

1
2
3
4
5
# 列出启动的 session
vncserver -list 

# kill session
vncserver -kill :<session-no>  # 比如 :2

参考

linux►distro►ubuntu

ubuntu user manager 

1
2
3
4
5
6
7
8
9
10
11
# 添加 group
$ groupadd {groupname}

# 添加用户
$ useradd -d /home/{username} -m -s /bin/bash -g {username} {groupname}

# 添加 sudo
$ sudo usermod -aG sudo {username}

# 设置密码
$ sudo passwd {username}
1
2
3
4
5
NEW_USER=wii
sudo groupadd $NEW_USER
sudo useradd -d /data/$NEW_USER -m -s /bin/bash -g $NEW_USER $NEW_USER
sudo usermod -aG sudo $NEW_USER
sudo passwd $NEW_USER
1
2
3
4
groupadd zhenkai.sun
useradd -d /data/zhenkai.sun -m -s /bin/bash -g zhenkai.sun zhenkai.sun
sudo usermod -aG sudo zhenkai.sun
sudo passwd zhenkai.sun
coding►java►notes

java lock example

场景

等待条件发生

比如，有这样一个场景，在 spring boot 工程里面，有一个 controller，他会接受数据并把数据写入 kafka，然后返回写入 kafka 的结果。在调用 send 方法后，会得到一个 ListenableFuture 对象，这个对象可以传入 callback 对象，这是一个异步的过程，需要等待回调执行后，才能将结果返回给客户端。

我们就需要一种机制等待回调事件，这里用的模式如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
Object mutex = new Object();

onSuccess/onFailure[callback] {
    synchronized (mutex) {
        // 发送消息成功后, 唤醒在等待的线程
        mutex.notify(); 
    }
}

// 程序会先走到这里, 并等待 mutex 唤醒
synchronized (mutex) {
    mutex.wait();
}

return ...;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@Controller
@RequestMapping("kafka")
public class KafkaController {
    static class Keeper {
        String result;
    }

    @Resource
    KafkaTemplate<String, String> template;

    @RequestMapping(value = "put", method = RequestMethod.POST,
            produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public ResponseEntity<String> check(@RequestBody Message message) throws InterruptedException {
        ListenableFuture<SendResult<String, String>> f = template.send("example", message.getKey(), message.getMessage());
        Object mutex = new Object();
        final Keeper keeper = new Keeper();
        keeper.result = "unknown";
        f.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(@SuppressWarnings("NullableProblems") Throwable throwable) {
                keeper.result = "send message failed";
                synchronized (mutex) {
                    mutex.notify();
                }
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                keeper.result = "send message success";
                synchronized (mutex) {
                    mutex.notify();
                }            }
        });

        synchronized (mutex) {
            mutex.wait();
        }

        return ResponseEntity.ok(keeper.result);
    }
}
架构►mq►kafka

kafka attentions

  • 传输大文件
  • 每个partition的一个消息只能被一个consumer group中的一个consumer消费
  • consumer group 中的一个consumer可以消费多个partition
  • 一个consumer group 中的cunsomer processes必须小于partition的数量
  • each message is delivered exactly to one consumer from a group (with a same group id)
架构►mq►kafka

kafka commend line tools

[toc]

list consumer groups

1
2
# 列出所有 consumer group
./bin/kafka-consumer-groups.sh --bootstrap-server $bootstrap_servers --list

list topic

1
2
3
4
5
6
7
8
9
10
$ ./bin/kafka-topics.sh --list --zookeeper server.zk
__consumer_offsets
demo_kafka_topic_1
model_diff_update_111
model_diff_update_156
model_diff_update_785
model_diff_update_802

# 示例
$ ./bin/kafka-topics.sh --list --zookeeper 127.0.0.1:2181

清空 topic

1
kafka-configs.sh --zookeeper <zkhost>:2181 --alter --entity-type topics --entity-name <topic name> --add-config retention.ms=1000

查看 topic 信息

1
2
3
$ ./bin/kafka-topics.sh --zookeeper c3cloudsrv.zk.hadoop.srv:11000/kafka/c3cloudsrv-feeds --describe --topic model_diff_update_111
Topic:model_diff_update_111	PartitionCount:1	ReplicationFactor:1	Configs:retention.ms=300000
	Topic: model_diff_update_111	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
1
$ kafka-topics --zookeeper 127.0.0.1:2181 --describe --topic <topic-name>

消费

1
2
3
4
5
$ ./bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic <topic> --from-beginning
$ ./bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic <topic> # 默认最新消息开始消费

# oom 异常, 设置 KAFKA_HEAP_OPTS="-Xms512m -Xmx1g"
$ env KAFKA_HEAP_OPTS="-Xms512m -Xmx1g" ./bin/kafka-console-consumer.sh --bootstrap-server <bootstrap-servers> --topic <topic> --from-beginning

指定配置文件

kafka-auth.properties

1
2
3
4
5
security.protocol        = "ssl"
ssl.ca.location          = "ca_cert.pem"
ssl.certificate.location = "client_cert.pem"
ssl.key.location         = "client_cert_key.pem"
ssl.key.password         = "password"
1
2
3
4
5
# 指定配置文件
## consumer
... --consumer.config kafka-auth.properties
## producer
... --producer.config kafka-auth.properties

参考

环境变量指定认证

1
2
3
4
5
6
7
8
9
10
CONNECT_CONSUMER_SECURITY_PROTOCOL: SASL_SSL
CONNECT_CONSUMER_SASL_KERBEROS_SERVICE_NAME: "kafka"
CONNECT_CONSUMER_SASL_JAAS_CONFIG: com.sun.security.auth.module.Krb5LoginModule required \
                            useKeyTab=true \
                            storeKey=true \
                            keyTab="/etc/kafka-connect/secrets/kafka-connect.keytab" \
                            principal="<principal>;
CONNECT_CONSUMER_SASL_MECHANISM: GSSAPI
CONNECT_CONSUMER_SSL_TRUSTSTORE_LOCATION: <path_to_truststore.jks>
CONNECT_CONSUMER_SSL_TRUSTSTORE_PASSWORD: <PWD>

发送消息

1
$ /bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <topic>

删除 topic

1
$ ./bin/kafka-topics --delete --zookeeper 127.0.0.1:2181 --topic <topic-name>
架构►mq►kafka

kafka notes

基础

  • 架构

    • 介绍
    • Rebalance
      • Why
        • 是Kafka集群的一个保护设置，用于剔除掉无法消费或过慢的消费者
        • 负载均衡
      • When
        • new consumer
        • consumer offline / exit / dead / unsubscribe
        • 消费者订阅的topic出现分区数量变化
      • 影响
        • 重复消费
        • Rebalance扩散到整个ConsumerGroup，一个Consumer的退出，导致Group进行Rebalance，影响面大
        • 频繁的Rebalance导致重复消费及Rebalance占用大量时间
        • 数据不能及时消费，会累计lag（消费滞后），在Kafka的TTL之后会丢弃数据
      • improve
        • 关闭 auto commit，手动管理offset和心跳
      • Rebalance Listener
      • more

  • 名词

    • Broker
      • 消息中间件处理节点
      • 一个Kafka节点就是一个broker
      • 一个或多个Kafka节点组成Kafka集群
    • Topic
      • Kafka根据Topic对消息进行归类
      • 发布到Kafka集群的每条消息都需要指定一个topic
    • Producer
      • 参数
        • Topic
        • Partition（Optional）
        • Key（Optional）
        • Value
    • Consumer
      • 一个Consumer可以消费一个或多个partition
    • ConsumerGroup
      • 每个Consumer属于一个特定的ConsumerGroup
      • 一条消息可以发送到多个不同的ConsumerGroup
      • 一个ConsumerGroup中的一条消息只被一个Consumer处理
      • 仅是用来对消费者进行分组来消费topic的消息
    • Partition
      • 物理上的概念
      • 一个topic的信息可以划分到多个partition
      • 每个partition内部是有序的
      • 每个partition在存储层面是append log文件
      • 顺序写磁盘，效率非常高，这是Kafka高吞吐量的重要保证
      • partition是broker属性，不影响producer
    • Segment
      • partition细分的物理概念
      • 包括：
        • .index文件
        • .log文件
      • Ref
    • Offset
      • 发布到partition的消息被追加到log文件的尾部
      • 每条消息在partition文件中的位置成为offset
      • offset是一个整形数字，唯一标记一条消息
    • Leader & Follower
      • 为了提高消息可靠性，Kafka为每个topic的partition设置N个副本
      • N 副本中的一个选举为Leader，其他为Follower
      • Leader处理partition的所有读写请求，follower定期复制leader上的数据
      • 负责维护和跟踪ISR（副本同步队列）中所有follower滞后的状态
      • producer发送一条消息到broker后，leader写入消息并复制到所有follower

  • 如何提高可靠性

    • 通过request.required.acks参数设置数据可靠性级别
      • 1：producer接受到leader成功收到数据并得到确认后发送下一条数据
        • 如果leader宕机，消息未同步到follower，可能会造成数据丢失
      • 0：producer无需等待来自broker的确认而继续发送下一条消息，可靠性最低
      • -1：producer等待ISR中所有follower都确认接收到数据才算一次发送成功，可靠性最高
        • 如果设置副本为1，也就是说只有leader，此时和设置request.required.acks等效，如果leader宕机，也会发生数据丢失
    • 保证高可靠性
      • topic的配置：replication.factor>=3,即副本数至少是个;2<=min.insync.replicas<=replication.factor
      • broker的配置：leader的选举条件unclean.leader.election.enable=false
      • producer的配置：request.required.acks=-1(all)，producer.type=sync

  • 应用

    • 日志收集
    • 业务数据收集
    • page view
    • …

  • Ref

读取消息示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
val props = new Properties()
props.setProperty("group.id", "-")
props.setProperty("bootstrap.servers", "-")
props.setProperty("auto.offset.reset", "-")
props.put("key.deserializer", classOf[StringDeserializer])
props.put("value.deserializer", classOf[StringDeserializer])
props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor")
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(List(topic))
while (true) {
  val results = consumer.poll(2000)
  for (record <- results.iterator()) {
    print(s"${record.key()} - ${record.value()}")
  }
}
架构►mq►kafka

kafka QA

  • consumer 和 consumer group的关系？
    • 一个用户属于一个 consumer group
  • consumer group 和 topic 的关系？
    • consumer group保存自己的offset
    • 不同consumer group消费同一topic时，会消费同样的内容，各个group保存自己的offset
  • topic 和 partition 的关系？
    • topic 和 partition 不在一个抽象层次
    • 一个 topic 的消息会被划分到多个partition（如果partion数量被设定 > 1）
  • consumer group 和 partition 的关系？
    • 同样，consumer group 和 partition 也不在一个抽象层次
  • partition 和 replication 的关系？
    • 没关系
  • 怎么指定partition？
    • 不允许使用 producer api 设置partition数量
    • 每个topic的partition数量根据配置文件中的num.partitioins 指定
架构►mq►kafka

kafka offset

定义

偏移量是，在一个分区内下一条需要发送给消费者的消息位置。Kafka包括两种类型的offset：

  • Current offset
    • 当前偏移量是指向Kafka在最近一次轮询中已发送给消费者的最后一条记录的指针。所以Current Offset消费者不会获取到相同的记录
  • Committed offset
    • Commited Offset 是消费者已确认处理的位置

总结

  • Current offset：Send Records：用来避免向同一个消费者发送相同的数据
  • Committed offset：Processed Records：避免在分区平衡的状况下，向新的消费者发送相同的记录

Commited Offset在分区平衡的情况下至关重要。分区平衡的情况下，新的消费者被分配到新的分区时，Committed Offset可以解决从哪里开始、哪些记录已经被消费的问题。

Commit an offset

Current offset 和 committed offset由Kafka管理。提交一个offset的方式有两种：

  • Auto commit
  • Manual-commit

Auto commit

通过两个属性来控制Auto-Commit：

  • enable.auto.commit
    • 默认为true，所有 auto-commit 默认开启
  • auto.commit.interval.ms
    • 定义auto-commit时间间隔

auto-commit的一个问题是，在提交之前，数据可能会被其他消费者处理。这种情况下，没办法彻底避免消息被重复处理。

Manual commit

可以使用manual-commit的方式解决auto-commit的问题。manual-commit由两种方式：

  • Commit Sync
  • Commit Async

一个简单示例。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import java.util.*;
import java.io.*;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ManualConsumer{

    public static void main(String[] args) throws Exception{

        String topicName = "SupplierTopic";
        String groupName = "SupplierTopicGroup";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093");
        props.put("group.id", groupName);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "SupplierDeserializer");
        props.put("enable.auto.commit", "false");

        KafkaConsumer<String, Supplier> consumer = null;

        try {
            consumer = new KafkaConsumer<>(props);
            consumer.subscribe(Arrays.asList(topicName));

            while (true){
                ConsumerRecords<String, Supplier> records = consumer.poll(100);
                for (ConsumerRecord<String, Supplier>record : records){
                    System.out.println("Supplier id= " + String.valueOf(record.value().getID()) + 
                                       " Supplier  Name = " + record.value().getName() + 
                                       " Supplier Start Date = " + record.value().getStartDate().toString());
                }
                consumer.commitAsync();
            }
        }catch(Exception ex){
            ex.printStackTrace();
        }finally{
            consumer.commitSync();
            consumer.close();
        }
    }
}

这里：

  • 使用异步提交
  • 当出现异常或退出时，使用同步提交

Offset backend storage

Offset记录位置是根据Kafka broker版本和Kafka client版本决定。

Kafka version\Kafka deriver version <0.9 >=0.9
<0.9 Offset Storage: Zookeeper Offset Storage: Zookeeper
>=0.9 Offset Storage: Zookeeper Offset Storage: Kafka

如果Broker存储Offset，处理方式：

  • Kafka把Offset作为Message存储在topic __consumer_offsets 中
  • 每个consumer定期向这个topic 提交Message，Message包括
    • current offset
    • consumer group
    • partition number
    • topic

在kafka 0.11.x 版本配合 cppkafka client，可以断定，offset 后端存储是kafka broker进行管理，依据：

  • 可以查询到 __consumer_offsets 主题，且有offsets信息
1
2
3
$ ./kafka-console-consumer.sh --consumer.config /tmp/consumer.config --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --topic __consumer_offsets --zookeeper *:2181/feeds/infra/feeds-kafka-srv --from-beginning
[consumer,test,0]::[OffsetMetadata[332,NO_METADATA],CommitTime 1537347025096,ExpirationTime 1537433425096]
...
  • 对应zookeeper节点中没有对应offsets信息
    • 路径格式：/consumers/{CONSUMER_GROUP_ID}/offsets/{TOPIC_NAME}/{PARTITION_NUMBER}

Ref

参考