Kafka 集群是如何选择 leader,你知道吗?
kafka
集群是由多个broker
节点组成,这里面包含了许多的知识点,以下的这些问题你都知道吗?
你知道
topic
的分区leader
是怎么选举的吗?你知道
zookeeper
中存储了kafka
的什么信息吗?起到什么做呢?你知道
kafka
消息文件是怎么存储的吗?如果
kafka
中leader
节点或者follower
节点发生故障,消息会丢失吗?如何保证消息的一致性和可靠性呢?
如果你对这些问题比较模糊的话,那么很有必要看看本文,去了解以下kafka
的核心设计,本文主要基于kafka
3.x 版本讲解。
kafka broker 核心机制
kafka 集群整体架构
kafka 集群是由多个kafka broker
通过连同一个zookeeper
组成,那么他们是如何协同工作对外提供服务的呢?zookeeper
中又存储了什么信息呢?
![](https://static001.geekbang.org/infoq/73/7305ae63f49e8f4723869e1d558cb91c.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
kafka broker
启动后,会在zookeeper
的/brokers/ids
路径下注册。同时,其中一个
broker
会被选举为控制器(Kafka Controller
)。选举规则也很简单,谁先注册到zookeeper
中的/controller
节点,谁就是控制器。Controller
主要负责管理整个集群中所有分区和副本的状态。Kafka Controller
会进行Leader
选择,比如上图中针对TopicA
中的 0 号分区,选择broker0
作为Leader
, 然后会将选择的节点信息注册到zookeeper
的/brokers/topics
路径下,记录谁是Leader
,有哪些服务器可用。被选举为
Leader
的topic
分区提供对外的读写服务。为什么只有Leader
节点提供读写服务,而不是设计成主从方式,Follower
提供读服务呢?
为了保证数据的一致性,因为消息同步延迟,可能导致消费者从不同节点读取导致不一致。
kafka 设计目的是分布式日志系统,不是一个读多写少的场景,kafka 的读写基本是对等的。
主从方式的话带来设计上的复杂度。
kafka leader 选举机制
那么问题来了,kafka
中topic
分区是如何选择leader
的呢?为了更好的阐述,我们先来理解下面 3 个概念。
ISR
:表示和Leader
保持同步的Follower
集合。如果Follower
长时间未向Leader
发送通信请求或同步数据,则该Follower
将被踢出ISR
。该时间阈值由replica.lag.time.max.ms
参数设定,默认30s
。Leader
发生故障之后,就会从ISR
中选举新的Leader
。OSR
:表示Follower
与Leader
副本同步时,延迟过多的副本。AR
:指的是分区中的所有副本,所以AR = ISR + OSR
。
Kafka Controller
选举Leader
的规则:在isr
队列中存活为前提,按照AR
中排在前面的优先。例如ar[1,0,2]
, isr [1,0,2]
,那么leader
就会按照 1,0,2 的顺序轮询。而 AR 中的这个顺序kafka
会进行打散,分摊kafka broker
的压力。
当运行中的控制器突然宕机或意外终止时,Kafka
通过监听zookeeper
能够快速地感知到,并立即启用备用控制器来代替之前失败的控制器。这个过程就被称为 Failover
,该过程是自动完成的,无需你手动干预。
![](https://static001.geekbang.org/infoq/63/63cc16223bb7ea704108bdb5a801a3cf.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
开始的时候,Broker 0
是控制器。当 Broker 0
宕机后,ZooKeeper
通过`` Watch 机制感知到并删除了 /controller
临时节点。之后,所有存活的 Broker
开始竞选新的控制器身份。Broker 3
最终赢得了选举,成功地在 ZooKeeper
上重建了 /controller
节点。之后,Broker 3
会从 ZooKeeper
中读取集群元数据信息,并初始化到自己的缓存中,后面就有Broker 3
来接管选择Leader
的功能了。
Leader 和 Follower 故障处理机制
如果topic
分区的leader
和follower
发生了故障,那么对于数据的一致性和可靠性会有什么样的影响呢?
![](https://static001.geekbang.org/infoq/76/7653ab397c97acc8874e7faf7f981f00.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
LEO(Log End Offset)
:每个副本的最后一个offset
,LEO
就是最新的offset
+ 1。HW(High Watermark)
:水位线,所有副本中最小的LEO
,消费者只能看到这个水位线左边的消息,从而保证数据的一致性。
上图所示,如果follower
发生故障怎么办?
Follower
发生故障后会被临时踢出ISR
队列。这个期间
Leader
和Follower
继续接收数据。待该
Follower
恢复后,Follower
会读取本地磁盘记录的上次的HW
,并将log
文件高于HW
的部分截取掉,从HW
开始向Leader
进行同步。等该
Follower
的LEO
大于等于该Partition
的HW
,即Follower
追上Leader
之后,就可以重新加入 ISR 了。
如果leader
发生故障怎么办?
![](https://static001.geekbang.org/infoq/23/232980302984f32dda94996bdd0f6adf.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
Leader
发生故障之后,会从ISR
中选出一个新的Leader
为保证多个副本之间的数据一致性,其余的
Follower
会先将各自的log
文件高于HW
的部分截掉,然后从新的Leader
同步数据。
所以为了让kafka broker
保证消息的可靠性和一致性,我们要做如下的配置:
设置 生产者
producer
的配置acks=all
或者-1。leader
在返回确认或错误响应之前,会等待所有副本收到悄息,需要配合min.insync.replicas
配置使用。这样就意味着leader
和follower
的LEO
对齐。设置
topic
的配置replication.factor>=3
副本大于 3 个,并且min.insync.replicas>=2
表示至少两个副本应答。设置
broker
配置unclean.leader.election.enable=false
,默认也是 false,表示不对落后leader
很多的follower
也就是非ISR
队列中的副本选择为Leader
, 这样可以避免数据丢失和数据 不一致,但是可用性会降低。
Leader Partition 负载平衡
正常情况下,Kafka
本身会自动把Leader Partition
均匀分散在各个机器上,来保证每台机器的读写吞吐量都是均匀的。但是如果某些broker
宕机,会导致 Leader Partition 过于集中在其他少部分几台broker
上,这会导致少数几台broker
的读写请求压力过高,其他宕机的 broker 重启之后都是follower partition
,读写请求很低,造成集群负载不均衡。那么该如何负载平衡呢?
自动负载均衡
通过broker
配置设置自动负载均衡。
auto.leader.rebalance.enable
:默认是true
。 自动Leader Partition
平衡。生产环境中,leader
重选举的代价比较大,可能会带来性能影响,建议设置为 false 关闭。leader.imbalance.per.broker.percentage
:默认是10%
。每个broker
允许的不平衡的leader
的比率。如果每个broker
超过了这个值,控制器会触发leader
的平衡。leader.imbalance.check.interval.seconds
:默认值300
秒。检查leader
负载是否平衡的间隔时间。
手动负载均衡
对所有
topic
进行负载均衡
对指定
topic
负载均衡
kafka 的存储机制
kafka 消息最终会存储到磁盘文件中,那么是如何存储的呢?清理策略是什么呢?
![](https://static001.geekbang.org/infoq/ca/ca1934f690a1a424b42dc4d2b42199cf.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
一个topic
分为多个partition
,每个 partition 对应于一个log
文件,为防止 log 文件过大导致数据定位效率低下,Kafka 采取了分片和索引机制,每个partition
分为多个segment
。每个segment
包括:“.index
”文件、“.log
”文件和.timeindex
等文件,Producer
生产的数据会被不断追加到该 log 文件末端。
![](https://static001.geekbang.org/infoq/93/931fc1c310570100b9fbfe6be3a0fbff.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
上图中 t1 即为一个topic
的名称,而“t1-0/t1-1”则表明这个目录是 t1 这个topic
的哪个partition
。
![](https://static001.geekbang.org/infoq/a3/a3cb563a61889a5a9b333bf93a2507a1.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
kafka 中的索引文件以稀疏索引(sparseindex
)的方式构造消息的索引,如下图所示:
![](https://static001.geekbang.org/infoq/03/03a0cba1393ba1e4be50cf9540c0d480.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
根据目标
offset
定位segment
文件找到小于等于目标
offset
的最大offset
对应的索引项定位到
log
文件向下遍历找到目标
Record
注意:index 为稀疏索引,大约每往
log
文件写入4kb
数据,会往index
文件写入一条索引。通过参数log.index.interval.bytes
控制,默认4kb
。
那 kafka 中磁盘文件保存多久呢?
kafka 中默认的日志保存时间为 7 天,可以通过调整如下参数修改保存时间。
log.retention.hours
,最低优先级小时,默认 7 天。log.retention.minutes
,分钟。log.retention.ms
,最高优先级毫秒。log.retention.check.interval.ms
,负责设置检查周期,默认 5 分钟。
kafka broker 重要参数
前面讲解了kafka broker
中的核心机制,我们再来看下重要的配置参数。
首先来说下 kafka 服务端配置属性Update Mode
的作用:
![](https://static001.geekbang.org/infoq/98/98d8f99639924ddceb2c74cd226eaa4c.webp?x-oss-process=image%2Fresize%2Cp_80%2Fformat%2Cpng)
read-only
。被标记为read-only
的参数和原来的参数行为一样,只有重启Broker
,才能令修改生效。per-broker
。被标记为 per-broker 的参数属于动态参数,修改它之后,无需重启就会在对应的broker
上生效。cluster-wide
。被标记为cluster-wide
的参数也属于动态参数,修改它之后,会在整个集群范围内生效,也就是说,对所有broker
都生效。也可以为具体的broker
修改cluster-wide
参数。
Broker 重要参数
![](https://static001.geekbang.org/infoq/48/48036ea1a32f140f5a0717a97789b7c4.png)
总结
Kafka
集群的分区多副本架构是 Kafka
可靠性保证的核心,把消息写入多个副本可以使 Kafka
在发生崩溃时仍能保证消息的持久性。本文围绕这样的核心架构讲解了其中的一些核心机制,包括 Leader 的选举、消息的存储机制等等。
作者:JAVA 旭阳
链接:https://juejin.cn/post/7238604003599695928
来源:稀土掘金
评论