Kafka 基本原理整理

Posted 2020-11-07

Kafka

体系架构

• Producers
• push message
• Brokers
• contain messages
• Consumers
• pull message
• Zookeeper Cluster
• manage kafka cluster config
• select leader
• rebalance consumer group

存储机制

Topic

• 一类消息，一个消息主题

Partition

• 一个 Topic 有多个 partition
• 每个 partition 为每个 consumer group 维护了一个 逻辑offset
• 一个 partition 在存储中对应一个目录，分成多个 segments
• 可以对 partition 做备份
• partition 内的消息是有序的

Segment

• 每个 segments 对应一个 .log 文件和一个 .index 文件，在磁盘中顺序存储消息（比随机写内存效率要高）
• .index 和 .log 文件的命名方式是以逻辑 offset 命名的，第一个 00000000.log（20位） 00000000.index，后面的可能是 00123123.log 00123123.index
• 分割是由配置决定的，分割的时间 或者 分割的大小
• .index 相当于一个索引文件，文件每行存放了一个局部 offset 和它在 .log 文件的偏移量，offset 是稀疏的，并非连贯的，用以减少 .index 存储
• .log 文件有自己的格式，会记录一些元素，以及各个元素的偏移量，所以只要找到偏移量，就能遍历下面的条数，找到对应的数据，二分查找（这个不太确定）
• 总结起来，根据 offset 查找对应的 .index，然后对 .index 进行二分查找，确定偏移量，然后从 .log 文件中根据偏移量查找每行数据，直到找到 offset 对应的那一条数据

高可靠 Tips：这种顺序存储机制保证了快速读写（顺序存储，索引），负载均衡（partition），快速过期删除（segment），以及容灾备份（replica）

Kafka 没有一个缓存机制，每次都要访问文件吗？

复制和同步

• HW HighWatermark
• 每个 partition（包括 partition 副本） 都会有一个 HW
• 这个 HW 决定了能读取的最大偏移量
• LEO LogEndOffset
• 每个 partition 也都会有一个 LEO
• 这个是真正的消息记录的截止位置
• ISR In-Sync-Replicas
• 每个 partition leader 维护了一个 ISR 列表，即副本同步队列，保存了 partition follower
• 如果 follower 过慢，则可能会被从列表删除
• replica.lag.time.max.ms
• replica.lag.max.messages
• OSR Out-Sync-Replicas
• 从 ISR 中被移除的会放在 OSR 中
• AR Assigned-Replicas
• OSR + ISR

kafka 的复制机制不是完全同步的，也不是单纯的异步复制

• 同步复制降低了吞吐率
• 异步复制可能会丢失数据
• ISR 可以很好的均衡上述两点

ISR 的信息都会反馈到 zookeeper 上，有两个地方会维护这个信息

• broker controller
• 负责管理 partition 和 replica 状态
• 重新分配 partition
• LeaderSelector 选举新的 partition leader，ISR，leader_epoch，controller_epoch
• 把相关消息推送给所有 replica
• partition leader
• 检查 ISR，更新 zk

ISR 包括了 partition leader 自身

replication=3 表示 算主有3个

数据可靠性和持久性

ack

• 1 leader 确认则可直接发送下一条数据
• 0 不用等 leader 确认
• -1 ISR 所有 follower 确认
• 当配置了 min.insync.replicas 这个参数，会发挥其功效，就是 至少这个数的 ISR 中的 follower 确定后才算提交成功，否则返回异常

担心切换 leader 时数据丢失，因为 leader 是否会选择最新的，而不是随机选的

在 -1 的情况下：

• kafka 同步，replication.factor >= 2 && min.insync.replicas >= 2，这种情况不会丢失数据
• 如果 kafka broker 宕机， ISR 中的 follower 没有全部同步，而返回了异常，这时候如果选择了已经同步的 follower，会造成数据重复

恢复后同步

• 要确保一致性
• 不会从 LEO 开始
• 会从 HW 开始
• 因为可能 LEO 可能没有同步完就 down 掉了，所以从 LEO 开始会多数据，造成不一致

leader 选举

• 不是少数服从多数，raft 这种是，zk 是，这种方式需要大量的副本
• 大量的副本会在大数据量下导致性能的急剧下降
• 很少在需要大量数据的系统中使用
• 常见的选举算法
• Zab
• Raft
• Paxos
• PacificA
• Viewstamped Replication

容错处理

如果某一个partition的所有replica都挂了，就无法保证数据不丢失了。这种情况下有两种可行的方案：

• 等待ISR中任意一个replica“活”过来，并且选它作为leader
• 选择第一个“活”过来的replica（并不一定是在ISR中）作为leader

默认会采用第二种

对于 某个 broker down 掉，可能导致服务不可用（可读不可写，ack=-1，replicas>1），这时候需要调整 min.insync.replicas = 1

Procduer 发送方式

• producer.type=sync

• 可靠

• producer.type=async
• batch 方式

• 有数据丢失的危险

如果 producer 网络出现问题，没有收到 ack，也会重试，所以会出现 at least once；如果 consumer 设置了自动提交，那么在 producer 没出问题的前提下，是 exactly once。如果手动提交，在消费结束后提交，就是 at least once，如果在之前，就是 at most once，因为可能消费失败。所以为了保证 exactly once，需要消费后手动提交，并加入去重机制。

总结

要保证数据写入到Kafka是安全的，高可靠的，需要如下的配置：

• topic的配置：replication.factor>=3,即副本数至少是3个；2<=min.insync.replicas<=replication.factor
• broker的配置：leader的选举条件unclean.leader.election.enable=false（ISR中选取Leader）
• producer的配置：request.required.acks=-1(all)，producer.type=sync

测试表现：

• 当acks=-1时，Kafka发送端的TPS受限于topic的副本数量（ISR中），副本越多TPS越低；
• acks=0时，TPS最高，其次为1，最差为-1，即TPS：acks_0 > acks_1 > acks_-1；
• min.insync.replicas参数不影响TPS；
• partition的不同会影响TPS，随着partition的个数的增长TPS会有所增长，但并不是一直成正比关系，到达一定临界值时，partition数量的增加反而会使TPS略微降低；