kafka个人笔记

大部分内容源于https://segmentfault.com/a/1190000038173886, 本人手敲一边加强印象方便复习

消息系统的作用

解耦
冗余
扩展性
灵活性（峰值处理
可恢复
顺序保证
缓冲
异步

架构

发布消息

producer用push发布到broker，消息被append到partition，顺序写磁盘

消息路由

//构造函数
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
     if (topic == null)
          throw new IllegalArgumentException("Topic cannot be null");
     if (timestamp != null && timestamp < 0)
          throw new IllegalArgumentException("Invalid timestamp " + timestamp);
     this.topic = topic;
     this.partition = partition;
     this.key = key;
     this.value = value;
     this.timestamp = timestamp;
}


private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
     Integer partition = record.partition();
     if (partition != null) {//指定了 partition 则直接使用
          List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
          int lastPartition = partitions.size() - 1;
          if (partition < 0 || partition > lastPartition) {
               throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
          }
          return partition;
     }//否则使用 key 计算
     return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
     List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
     int numPartitions = partitions.size();
     if (keyBytes == null) {//轮询
          int nextValue = counter.getAndIncrement();
          List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
          if (availablePartitions.size() > 0) {
               int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
               return availablePartitions.get(part).partition();
          } else {
               return DefaultPartitioner.toPositive(nextValue) % numPartitions;
          }
     } else {
          //对 keyBytes 进行 hash 选出一个 patition
          return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
     }
}

写入流程

投递保证

    ① At most once 消息可能会丢，但绝不会重复传递

    ② At least one  消息绝不会丢，但可能会重复传递

    ③ Exactly once 每条消息肯定会被传输一次且仅传输一次，很多时候这是用户想要的

默认 at least one

接收消息的行为

at most once

读完消息先commit再处理消息。
若commit后未处理消息系统崩坏，下次重新开始工作无法读到已提交但未处理的消息

At least once

读完消息先处理再commit消费状态(保存offset)
若处理消息后未commit系统崩坏，重新工作的时候会处理未commit的消息（处理两次）

Exactly once 两阶段提交

协调offset和实际操作的输出。但由于许多输出系统不支持两阶段提交，更为通用的方式是将offset和操作输入存在同一个地方

（目前就high level API而言，offset是存于Zookeeper中的，无法存于HDFS，而low level API的offset是由自己去维护的，可以将之存于HDFS中）。

消息保存

topic分为多个partition，每个partition对应一个文件夹

无论消息是否被消费，kafka 都会保留所有消息。有两种策略可以删除旧数据

log.cleanup.policy=delete启用删除策略
直接删除，删除后的消息不可恢复。可配置以下两个策略：

清理超过指定时间清理： 
log.retention.hours=16

超过指定大小后，删除旧的消息：
log.retention.bytes=1073741824

topic的创造

删除 topic 的序列

kafka HA 高可用性

replica

同一个 partition 可能会有多个 replica —— erver.properties 配置中的 default.replication.factor=N

若没有replica，broker死机

引入replica，需要选取leader，leader与producer和consumer交互，其他replica与leader复制数据

分配规则

leader failover

partition 对应的 leader 宕机时，需要从 follower 中选举出新 leader

新的 leader 必须拥有旧 leader commit 过的所有消息

zookeeper 中（/brokers/…/state）动态维护了一个 ISR（in-sync replicas）。只有 ISR 里面的成员才能选为 leader。若有f个replica，partition可以保证f-1个replica失效情况下消息不丢失

failover方案

broker failover

controller failover

controller 宕机时会触发 controller failover

1. 读取并增加 Controller Epoch。
2. 在 reassignedPartitions Patch(/admin/reassign_partitions) 上注册 watcher。
3. 在 preferredReplicaElection Path(/admin/preferred_replica_election) 上注册 watcher。
4. 通过 partitionStateMachine 在 broker Topics Patch(/brokers/topics) 上注册 watcher。
5. 若 delete.topic.enable=true（默认值是 false），则 partitionStateMachine 在 Delete Topic Patch(/admin/delete_topics) 上注册 watcher。
6. 通过 replicaStateMachine在 Broker Ids Patch(/brokers/ids)上注册Watch。
7. 初始化 ControllerContext 对象，设置当前所有 topic，“活”着的 broker 列表，所有 partition 的 leader 及 ISR等。
8. 启动 replicaStateMachine 和 partitionStateMachine。
9. 将 brokerState 状态设置为 RunningAsController。
10. 将每个 partition 的 Leadership 信息发送给所有“活”着的 broker。
11. 若 auto.leader.rebalance.enable=true（默认值是true），则启动 partition-rebalance 线程。
12. 若 delete.topic.enable=true 且Delete Topic Patch(/admin/delete_topics)中有值，则删除相应的Topic。

消费

kafka 提供了两套 consumer API：

The high-level Consumer API
The SimpleConsumer API

consumer API

high-level提供kafka消费数据的抽象

使用high-level consumer API可以是多线程应用

if(消费线程 > partition){
	部分线程收不到消息
}
if(消费线程 < partition){
	有些线程收到多个partition消息
}

if(一个线程消费多个 patition){
	无法保证收到消息的顺序
}

** SimpleConsumer API**

适用以下情况

partition， offset， broker， leader不透明，需要自己管理

流程如下

consumer group
kafka分配单位是partition，consumer属于一个group
一个partition被一个group内的一个consumer消费（但是多个group可以同时消费这个partition）

实现离线处理与实时处理

消费方法

consumer用pull模式从broker读数据

push 模式很难适应消费速率不同的消费者

pull模式，consumer根据自己的能力消费信息

pull的优点

消费者递送保证

consumer 设置为 autocommit，consumer 一旦读到数据立即自动 commit（Exactly once

实际使用过程中，并不是consumer读完消息就结束了，还需要进一步处理。
处理和commit顺序决定了 consumer delivery guarantee

consumer rebalance

触发机制

算法如下

consumer调整了单个partition后，为了保证一致性，group内其他consumer也应触发balance

导致以下问题

herd effect

split brain