微服务架构之–消息队列Kafka图解最全知识点(3)

broker消息存储

• Kafka的消息以二进制的方式紧凑地存储，节省了很大空间
• 此外消息存在ByteBuffer而不是堆，这样broker进程挂掉时，数据不会丢失，同时避免了gc问题
• 通过零拷贝和顺序寻址，让消息存储和读取速度都非常快
• 处理fetch请求的时候通过zero-copy 加快速度

broker状态数据

• broker设计中，每台机器都保存了相同的状态数据。主要包括以下：
• controller所在的broker ID，即保存了当前集群中controller是哪台broker
• 集群中所有broker的信息：比如每台broker的ID、机架信息以及配置的若干组连接信息
• 集群中所有节点的信息：严格来说，它和上一个有些重复，不过此项是按照broker ID和监听器类型进行分组的。对于超大集群来说，使用这一项缓存可以快速地定位和查找给定节点信息，而无需遍历上一项中的内容，算是一个优化吧
• 集群中所有分区的信息：所谓分区信息指的是分区的leader、ISR和AR信息以及当前处于offline状态的副本集合。这部分数据按照topic-partitionID进行分组，可以快速地查找到每个分区的当前状态。(注：AR表示assigned replicas，即创建topic时为该分区分配的副本集合)

broker负载均衡

分区数量负载：各台broker的partition数量应该均匀

partition Replica分配算法如下：

1. 将所有Broker(假设共n个Broker)和待分配的Partition排序
2. 将第i个Partition分配到第(i mod n)个Broker上
3. 将第i个Partition的第j个Replica分配到第((i + j) mod n)个Broker上

容量大小负载：每台broker的硬盘占用大小应该均匀

在kafka1.1之前，Kafka能够保证各台broker上partition数量均匀，但由于每个partition内的消息数不同，可能存在不同硬盘之间内存占用差异大的情况。在Kafka1.1中增加了副本跨路径迁移功能kafka-reassign-partitions.sh，我们可以结合它和监控系统，实现自动化的负载均衡

Kafka高可用

在介绍kafka高可用之前先介绍几个概念

• 同步复制：要求所有能工作的Follower都复制完，这条消息才会被认为commit，这种复制方式极大的影响了吞吐率
• 异步复制：Follower异步的从Leader pull数据，data只要被Leader写入log认为已经commit，这种情况下如果Follower落后于Leader的比较多，如果Leader突然宕机，会丢失数据

Isr

Kafka结合同步复制和异步复制，使用ISR(与Partition Leader保持同步的Replica列表)的方式在确保数据不丢失和吞吐率之间做了平衡。Producer只需把消息发送到Partition Leader，Leader将消息写入本地Log。Follower则从Leader pull数据。Follower在收到该消息向Leader发送ACK。一旦Leader收到了ISR中所有Replica的ACK，该消息就被认为已经commit了，Leader将增加HW并且向Producer发送ACK。这样如果leader挂了，只要Isr中有一个replica存活，就不会丢数据。

Isr动态更新

Leader会跟踪ISR，如果ISR中一个Follower宕机，或者落后太多，Leader将把它从ISR中移除。这里所描述的“落后太多”指Follower复制的消息落后于Leader后的条数超过预定值(replica.lag.max.messages)或者Follower超过一定时间(replica.lag.time.max.ms)未向Leader发送fetch请求。

broker Nodes In Zookeeper

/brokers/topics/[topic]/partitions/[partition]/state 保存了topic-partition的leader和Isr等信息

Controller负责broker故障检查&&故障转移(fail/recover)

• Controller在Zookeeper上注册Watch，一旦有Broker宕机，其在Zookeeper对应的znode会自动被删除，Zookeeper会触发 Controller注册的watch，Controller读取最新的Broker信息
• Controller确定set_p，该集合包含了宕机的所有Broker上的所有Partition
• 对set_p中的每一个Partition，选举出新的leader、Isr，并更新结果。

3.1 从/brokers/topics/[topic]/partitions/[partition]/state读取该Partition当前的ISR

3.2 决定该Partition的新Leader和Isr。如果当前ISR中有至少一个Replica还幸存，则选择其中一个作为新Leader，新的ISR则包含当前ISR中所有幸存的Replica。否则选择该Partition中任意一个幸存的Replica作为新的Leader以及ISR(该场景下可能会有潜在的数据丢失)

3.3 更新Leader、ISR、leader_epoch、controller_epoch：写入/brokers/topics/[topic]/partitions/[partition]/state

直接通过RPC向set_p相关的Broker发送LeaderAndISRRequest命令。Controller可以在一个RPC操作中发送多个命令从而提高效率。

Controller挂掉

每个 broker 都会在 zookeeper 的临时节点 "/controller" 注册 watcher，当 controller 宕机时 "/controller" 会消失，触发broker的watch，每个 broker 都尝试创建新的 controller path，只有一个竞选成功并当选为 controller。

使用Kafka如何保证幂等性

不丢消息

• 首先kafka保证了对已提交消息的at least保证
• Sender有重试机制
• producer业务方在使用producer发送消息时，注册回调函数。在onError方法中重发消息
• consumer 拉取到消息后，处理完毕再commit，保证commit的消息一定被处理完毕

不重复

consumer拉取到消息先保存，commit成功后删除缓存数据

Kafka高性能

• partition提升了并发
• zero-copy
• 顺序写入
• 消息聚集batch
• 页缓存

业务方对 Kafka producer的优化

• 增大producer数量
• ack配置
• batch

（编辑：ASP站长网）