世界資訊：Kafka集群是如何選擇Leader，你知道嗎？

2023-05-30 08:59:19來源：JAVA旭陽

前言

kafka集群是由多個broker節(jié)點組成，這里面包含了許多的知識點，以下的這些問題你都知道嗎?

(資料圖)

你知道topic的分區(qū)leader是怎么選舉的嗎？你知道zookeeper中存儲了kafka的什么信息嗎？起到什么做呢？你知道kafka消息文件是怎么存儲的嗎？如果kafka中l(wèi)eader節(jié)點或者follower節(jié)點發(fā)生故障，消息會丟失嗎？如何保證消息的一致性和可靠性呢？

如果你對這些問題比較模糊的話，那么很有必要看看本文，去了解以下kafka的核心設(shè)計，本文主要基于kafka3.x版本講解。

kafka broker核心機制kafka集群整體架構(gòu)

kafka集群是由多個kafka broker通過連同一個zookeeper組成，那么他們是如何協(xié)同工作對外提供服務(wù)的呢？zookeeper中又存儲了什么信息呢？

kafka broker啟動后，會在zookeeper的/brokers/ids路徑下注冊。同時，其中一個broker會被選舉為控制器（Kafka Controller）。選舉規(guī)則也很簡單，誰先注冊到zookeeper中的/controller節(jié)點，誰就是控制器。Controller主要負責(zé)管理整個集群中所有分區(qū)和副本的狀態(tài)。Kafka Controller會進行Leader選擇，比如上圖中針對TopicA中的0號分區(qū)，選擇broker0作為Leader, 然后會將選擇的節(jié)點信息注冊到zookeeper的/brokers/topics路徑下，記錄誰是Leader，有哪些服務(wù)器可用。被選舉為Leader的topic分區(qū)提供對外的讀寫服務(wù)。為什么只有Leader節(jié)點提供讀寫服務(wù)，而不是設(shè)計成主從方式，F(xiàn)ollower提供讀服務(wù)呢？為了保證數(shù)據(jù)的一致性，因為消息同步延遲，可能導(dǎo)致消費者從不同節(jié)點讀取導(dǎo)致不一致。kafka設(shè)計目的是分布式日志系統(tǒng)，不是一個讀多寫少的場景，kafka的讀寫基本是對等的。主從方式的話帶來設(shè)計上的復(fù)雜度。kafka leader選舉機制

那么問題來了，kafka中topic分區(qū)是如何選擇leader的呢？為了更好的闡述，我們先來理解下面3個概念。

****ISR：表示和Leader保持同步的Follower集合。如果Follower長時間未向Leader發(fā)送通信請求或同步數(shù)據(jù)，則該Follower將被踢出ISR。該時間閾值由replica.lag.time.max.ms參數(shù)設(shè)定，默認30s。Leader發(fā)生故障之后，就會從ISR中選舉新的Leader。****OSR：表示Follower與Leader副本同步時，延遲過多的副本。****AR: 指的是分區(qū)中的所有副本，所以AR = ISR + OSR。

Kafka Controller選舉Leader的規(guī)則：在isr隊列中存活為前提，按照AR中排在前面的優(yōu)先。例如ar[1,0,2],isr [1，0，2]，那么leader就會按照1，0，2的順序輪詢。而AR中的這個順序kafka會進行打散，分?jǐn)俴afka broker的壓力。

當(dāng)運行中的控制器突然宕機或意外終止時，Kafka通過監(jiān)聽zookeeper能夠快速地感知到，并立即啟用備用控制器來代替之前失敗的控制器。這個過程就被稱為Failover，該過程是自動完成的，無需你手動干預(yù)。

開始的時候，Broker 0是控制器。當(dāng)Broker 0宕機后，ZooKeeper通過`` Watch 機制感知到并刪除了/controller臨時節(jié)點。之后，所有存活的Broker開始競選新的控制器身份。Broker 3最終贏得了選舉，成功地在ZooKeeper上重建了/controller節(jié)點。之后，Broker 3會從ZooKeeper中讀取集群元數(shù)據(jù)信息，并初始化到自己的緩存中，后面就有Broker 3來接管選擇Leader的功能了。

Leader 和 Follower 故障處理機制

如果topic分區(qū)的leader和follower發(fā)生了故障，那么對于數(shù)據(jù)的一致性和可靠性會有什么樣的影響呢？

LEO（Log End Offset）：每個副本的最后一個offset，LEO就是最新的offset+ 1。HW（High Watermark）：水位線，所有副本中最小的LEO，消費者只能看到這個水位線左邊的消息，從而保證數(shù)據(jù)的一致性。

上圖所示，如果follower發(fā)生故障怎么辦？

Follower發(fā)生故障后會被臨時踢出ISR隊列。這個期間Leader和Follower繼續(xù)接收數(shù)據(jù)。待該Follower恢復(fù)后，F(xiàn)ollower會讀取本地磁盤記錄的上次的HW，并將log文件高于HW的部分截取掉，從HW開始向Leader進行同步。等該Follower的LEO大于等于該Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了。

如果leader發(fā)生故障怎么辦？

Leader發(fā)生故障之后，會從ISR中選出一個新的Leader為保證多個副本之間的數(shù)據(jù)一致性，其余的Follower會先將各自的log文件高于HW的部分截掉，然后從新的Leader同步數(shù)據(jù)。

所以為了讓kafka broker保證消息的可靠性和一致性，我們要做如下的配置：

設(shè)置生產(chǎn)者producer的配置acks=all或者-1。leader在返回確認或錯誤響應(yīng)之前，會等待所有副本收到悄息，需要配合min.insync.replicas配置使用。這樣就意味著leader和follower的LEO對齊。設(shè)置topic的配置replication.factor>=3副本大于3個，并且min.insync.replicas>=2表示至少兩個副本應(yīng)答。設(shè)置broker配置unclean.leader.election.enable=false，默認也是false，表示不對落后leader很多的follower也就是非ISR隊列中的副本選擇為Leader, 這樣可以避免數(shù)據(jù)丟失和數(shù)據(jù) 不一致，但是可用性會降低。Leader Partition 負載平衡

正常情況下，Kafka本身會自動把Leader Partition均勻分散在各個機器上，來保證每臺機器的讀寫吞吐量都是均勻的。但是如果某些broker宕機，會導(dǎo)致Leader Partition過于集中在其他少部分幾臺broker上，這會導(dǎo)致少數(shù)幾臺broker的讀寫請求壓力過高，其他宕機的broker重啟之后都是follower partition，讀寫請求很低，造成集群負載不均衡。那么該如何負載平衡呢？

自動負載均衡

通過broker配置設(shè)置自動負載均衡。

auto.leader.rebalance.enable：默認是true。自動Leader Partition平衡。生產(chǎn)環(huán)境中，leader重選舉的代價比較大，可能會帶來性能影響，建議設(shè)置為 false 關(guān)閉。leader.imbalance.per.broker.percentage：默認是10%。每個broker允許的不平衡的leader的比率。如果每個broker超過了這個值，控制器會觸發(fā)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對應(yīng)于一個log文件，為防止log文件過大導(dǎo)致數(shù)據(jù)定位效率低下，Kafka采取了分片和索引機制，每個partition分為多個segment。每個segment包括：“.index”文件、“.log”文件和.timeindex等文件，Producer生產(chǎn)的數(shù)據(jù)會被不斷追加到該log文件末端。

上圖中t1即為一個topic的名稱，而“t1-0/t1-1”則表明這個目錄是t1這個topic的哪個partition。

kafka中的索引文件以稀疏索引（sparseindex）的方式構(gòu)造消息的索引，如下圖所示：

1.根據(jù)目標(biāo)offset定位segment文件

2.找到小于等于目標(biāo)offset的最大offset對應(yīng)的索引項

3.定位到log文件

4.向下遍歷找到目標(biāo)Record

注意：index為稀疏索引，大約每往log文件寫入4kb數(shù)據(jù)，會往index文件寫入一條索引。通過參數(shù)log.index.interval.bytes控制，默認4kb。

那kafka中磁盤文件保存多久呢？

kafka 中默認的日志保存時間為 7 天，可以通過調(diào)整如下參數(shù)修改保存時間。

log.retention.hours，最低優(yōu)先級小時，默認 7 天。log.retention.minutes，分鐘。log.retention.ms，最高優(yōu)先級毫秒。log.retention.check.interval.ms，負責(zé)設(shè)置檢查周期，默認 5 分鐘。kafka broker重要參數(shù)

前面講解了kafka broker中的核心機制，我們再來看下重要的配置參數(shù)。

首先來說下kafka服務(wù)端配置屬性Update Mode的作用：

read-only。被標(biāo)記為read-only的參數(shù)和原來的參數(shù)行為一樣，只有重啟Broker，才能令修改生效。per-broker。被標(biāo)記為 per-broker 的參數(shù)屬于動態(tài)參數(shù)，修改它之后，無需重啟就會在對應(yīng)的broker上生效。cluster-wide。被標(biāo)記為cluster-wide的參數(shù)也屬于動態(tài)參數(shù)，修改它之后，會在整個集群范圍內(nèi)生效，也就是說，對所有broker都生效。也可以為具體的broker修改cluster-wide參數(shù)。

Broker重要參數(shù)

參數(shù)名稱	描述
replica.lag.time.max.ms	ISR 中，如果 Follower 長時間未向 Leader 發(fā)送通信請求或同步數(shù)據(jù)，則該 Follower 將被踢出 ISR。該時間閾值，默認 30s。
auto.leader.rebalance.enable	默認是 true。自動 Leader Partition 平衡。
leader.imbalance.per.broker.percentage	默認是 10%。每個 broker 允許的不平衡的 leader的比率。如果每個 broker 超過了這個值，控制器會觸發(fā) leader 的平衡。
leader.imbalance.check.interval.seconds	默認值 300 秒。檢查 leader 負載是否平衡的間隔時間。
log.segment.bytes	Kafka 中 log 日志是分成一塊塊存儲的，此配置是指 log 日志劃分成塊的大小，默認值 1G。
log.index.interval.bytes	默認 4kb，kafka 里面每當(dāng)寫入了 4kb 大小的日志（.log），然后就往 index 文件里面記錄一個索引。
log.retention.hours	Kafka 中數(shù)據(jù)保存的時間，默認 7 天。
log.retention.minutes	Kafka 中數(shù)據(jù)保存的時間，分鐘級別，默認關(guān)閉。
log.retention.ms	Kafka 中數(shù)據(jù)保存的時間，毫秒級別，默認關(guān)閉。
log.retention.check.interval.ms	檢查數(shù)據(jù)是否保存超時的間隔，默認是 5 分鐘。
log.retention.bytes	默認等于-1，表示無窮大。超過設(shè)置的所有日志總大小，刪除最早的 segment。
log.cleanup.policy	默認是 delete，表示所有數(shù)據(jù)啟用刪除策略；如果設(shè)置值為 compact，表示所有數(shù)據(jù)啟用壓縮策略。
num.io.threads	默認是 8。負責(zé)寫磁盤的線程數(shù)。整個參數(shù)值要占總核數(shù)的 50%。
num.replica.fetchers	副本拉取線程數(shù)，這個參數(shù)占總核數(shù)的 50%的 1/3
num.network.threads	默認是 3。數(shù)據(jù)傳輸線程數(shù)，這個參數(shù)占總核數(shù)的50%的 2/3 。
log.flush.interval.messages	強制頁緩存刷寫到磁盤的條數(shù)，默認是 long 的最大值，9223372036854775807。一般不建議修改，交給系統(tǒng)自己管理。
log.flush.interval.ms	每隔多久，刷數(shù)據(jù)到磁盤，默認是 null。一般不建議修改，交給系統(tǒng)自己管理。