kafka设计原理

核心总控制器

• 集群中一般会有多个broker，其中一个会被选举为控制器
• 控制器会负责管理所有分区及其副本状态
• 分区leader出现故障，也有控制器负责选举新的leader
• 分区ISR信息发生变化，有控制器负责通知所有broker更新元信息
• broker增加分区时，同样由控制器负责让其他broker感知到

控制器选举

• 记录在zk中的controller临时节点中
• 每个broker启动，都会往zk发起 create controller 命令
• 原则上：谁先启动谁先创建成功，谁就是总控
• 总控挂了，其他节点发起重新选举
• 其他broker则会监听改controller节点，一旦临时节点消失，则会相互竞争再次创建该节点

Partition副本选举leader机制

• 总控制器监听着每个分区的leader信息
• 某个分区leader挂了后，总控再发起选举新的leader
• 控制器默认会从ISR列表中挑选第一个作为leader
• 若ISR列表中所有副本节点都挂了，则从列表外的副本挑选
• 提升了可用性，但数据可能不完整
• 因为不在ISR列表中的节点很可能是因为同步慢被剔除出去的节点

消费者消费消息的offset记录机制

• 每个消费者定期将自己的负责分区的offset提交给kafka内部topic
• key是consumerGroupId+topic+分区号
• value就是当前offset的值
• kafka会定期清理该topic里的消息，保留最新的那条
• 该topic默认给其分配50个分区
• 这样可以通过加机器的方式抗大并发

消费者Rebalance机制

• 消费组里的消费者数量有变化，会重新分配分区与消费者关系
• rebalance过程中，消费者无法从kafka消费消息
• 尽量避免在系统高峰期的重平衡发生
• rebalance的策略：range、round-robin、sticky
• range策略就是按照分区序号排序,系统默认

假设一个主题有10个分区(0-9)，现在有三个consumer消费：
range策略就是按照分区序号排序，假设 n＝分区数／消费者数量 = 3， m＝分区数%消费者数量 = 1，那么前 m 个消费者每个分配 n+1 个分区，后面的（消费者数量－m ）个消费者每个分配 n 个分区。

• round-robin策略就是轮询分配

比如分区0、3、6、9给一个consumer，分区1、4、7给一个consumer，分区2、5、8给一个consumer

• sticky策略初始时分配策略与round-robin类似，但是在rebalance的时候，需要保证如下两个原则。

1）分区的分配要尽可能均匀
2）分区的分配尽可能与上次分配的保持相同
两者发生冲突时，第一个目标优先于第二个目标，这样可以最大程度维持原来的分区分配的策略。

• rebanlance的过程主要有组协调器负责完成
• 组协调器选择方式：

consumer消费的offset要提交到__consumer_offsets的哪个分区，这个分区leader对应的broker就是这个consumer group的coordinator

发布消息机制剖析

• 每条消息被追加到分区中，是顺序写入磁盘，因此才可提供高吞吐量的消息写入

• 消息路由规则：

  • 指定了 patition，则直接使用
  • 未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition
  • patition 和 key 都未指定，使用轮询选出一个 patition

• 消息写入流程：

1. producer 先从 zookeeper 的 "/brokers/.../state" 节点找到该 partition 的 leader
2. producer 将消息发送给该 leader
3. leader 将消息写入本地 log
4. followers 从 leader pull 消息，写入本地 log 后 向leader 发送 ACK
5. leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 发送 ACK

HW与LEO

• HW俗称高水位，HighWatermark的缩写
• LEO(log-end-offset)
• 取一个partition对应的ISR中最小的LEO作为HW
• consumer最多只能消费到HW所在的位置
• 每个replica都有HW,leader和follower各自负责更新自己的HW的状态
• 对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费
• 思考其目的？均衡了数据不丢失以及吞吐率两个因素

日志分段存储

1. 分区的消息数据文件夹，以topic名称+分区号命名
2. 消息在分区内是分段(segment)存储，文件名以最开始的offset结尾
3. 一个段位的 log 文件最大为 1G
4. 避免单个文件过大，限制目的是为了提升读写性能

# 部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index文件，
# 如果要定位消息的offset会先在这个文件里快速定位，再去log文件里找具体消息
00000000000005367851.index
# 消息存储文件，主要存offset和消息体
00000000000005367851.log
# 消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset到timeindex文件，
# 如果需要按照时间来定位消息的offset，会先在这个文件里查找
00000000000005367851.timeindex

附图