前言
kafka
集群是由多个broker
节点组成,这里面包含了许多的知识点,以下的这些问题你都知道吗?
-
你知道
topic
的分区leader
是怎么选举的吗? -
你知道
zookeeper
中存储了kafka
的什么信息吗?起到什么做呢? -
你知道
kafka
消息文件是怎么存储的吗? -
如果
kafka
中leader
节点或者follower
节点发生故障,消息会丢失吗?如何保证消息的一致性和可靠性呢?
如果你对这些问题比较模糊的话,那么很有必要看看本文,去了解以下kafka
的核心设计,本文主要基于kafka
3.x 版本讲解。
kafka broker 核心机制
kafka 集群整体架构
kafka 集群是由多个kafka broker
通过连同一个zookeeper
组成,那么他们是如何协同工作对外提供服务的呢?zookeeper
中又存储了什么信息呢?
-
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
,该过程是自动完成的,无需你手动干预。
开始的时候,Broker 0
是控制器。当 Broker 0
宕机后,ZooKeeper
通过 Watch 机制感知到并删除了 /controller
临时节点。之后,所有存活的 Broker
开始竞选新的控制器身份。Broker 3
最终赢得了选举,成功地在 ZooKeeper
上重建了 /controller
节点。之后,Broker 3
会从 ZooKeeper
中读取集群元数据信息,并初始化到自己的缓存中,后面就有Broker 3
来接管选择Leader
的功能了。
Leader 和 Follower 故障处理机制
如果topic
分区的leader
和follower
发生了故障,那么对于数据的一致性和可靠性会有什么样的影响呢?
-
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
发生故障怎么办?
-
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
进行负载均衡
./bin/kafka-preferred-replica-election.sh --zookeeper hadoop16:2181,hadoop17:2181,hadoop18:2181/kafka08
复制代码
-
对指定
topic
负载均衡
cat topicPartitionList.json
{
"partitions":
[
{"topic":"test.example","partition": "0"}
]
}
复制代码
./bin/kafka-preferred-replica-election.sh --zookeeper hadoop16:2181,hadoop17:2181,hadoop18:2181/kafka08 --path-to-json-file topicPartitionList.json
复制代码
kafka 的存储机制
kafka 消息最终会存储到磁盘文件中,那么是如何存储的呢?清理策略是什么呢?
一个topic
分为多个partition
,每个 partition 对应于一个log
文件,为防止 log 文件过大导致数据定位效率低下,Kafka 采取了分片和索引机制,每个partition
分为多个segment
。每个segment
包括:“.index
”文件、“.log
”文件和.timeindex
等文件,Producer
生产的数据会被不断追加到该 log 文件末端。
上图中 t1 即为一个topic
的名称,而“t1-0/t1-1”则表明这个目录是 t1 这个topic
的哪个partition
。
kafka 中的索引文件以稀疏索引(sparseindex
)的方式构造消息的索引,如下图所示:
1.根据目标offset
定位segment
文件
2.找到小于等于目标offset
的最大offset
对应的索引项
3.定位到log
文件
4.向下遍历找到目标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
的作用:
-
read-only
。被标记为read-only
的参数和原来的参数行为一样,只有重启Broker
,才能令修改生效。 -
per-broker
。被标记为 per-broker 的参数属于动态参数,修改它之后,无需重启就会在对应的broker
上生效。 -
cluster-wide
。被标记为cluster-wide
的参数也属于动态参数,修改它之后,会在整个集群范围内生效,也就是说,对所有broker
都生效。也可以为具体的broker
修改cluster-wide
参数。
Broker 重要参数
总结
Kafka
集群的分区多副本架构是 Kafka
可靠性保证的核心,把消息写入多个副本可以使 Kafka
在发生崩溃时仍能保证消息的持久性。本文围绕这样的核心架构讲解了其中的一些核心机制,包括 Leader 的选举、消息的存储机制等等。