作者 | 蔡柱梁

審校 | 重樓

目錄

1. 前言
2. 可靠性分析
3. 副本設計
4. leader選舉機制
5. 日志同步機制

1 前言

本文里面涉及到較多基礎概念，如果忘記了，那么可以去看下《一文帶你快速入門kafka》。

對于一個消息中間件而言，可靠性是是至關重要的要素之一。不管是面試或者實際工作中，我們都不得不面對幾個問題：是幾個九？消息會不會丟失？如何保證冪等？如何順序消費？在這篇文章中，筆者會和大家一起去看 Kafka 是如何設計的。

2 可靠性分析

針對上面的幾個問題，Kafka 需要考慮包括并不限于以下問題：

• 可用性

– Kafka 支持分布式架構，實現了故障轉移，避免單點問題

如何避免腦裂問題（這個要了解 Kafka 的 leader 選舉機制）

– 多副本機制，支持容災備份

1. 數據一致性如何保證
2. 數據同步要如何實現

• 消息問題

– 生產者投遞消息

broker回復投遞成功，但是消息丟失了。出現這種情況，一般是以下幾種情況：

1. acks 配置成 0，生產者以為自己投遞成功了，但其實并沒有成功寫入 leader
2. 消息持久化在 leader 分區，但是沒有同步給 follower 就宕機了

這個問題也好解決：生產者可以在發送消息前，先將消息持久化。至多是用了存儲空間，現在磁盤空間可以說是最不值錢的了，而且我們還可以定期進行歸檔壓縮/刪除處理，問題不大。

– 消費者消費消息遇到消息丟失或者消息重復處理

• 消息丟失

消息丟失一般是以下這幾種情況：

1. 消費者拿到消息了，但是處理過程中發生異常
2. 消費者提交消費位移的設計不合理

針對這個問題，我們通常拿到消息會選擇將消息持久化在本地，然后再做消息處理，處理出問題也可以重復處理。這種設計滿足我們大多數場景，但是對于消息生產速度遠高于我們持久化速度的場景可能就不適用了，因為我們要考慮消息堆積問題。不管是這個問題，還是有些場景下無法找生產者重新投遞消息的問題，都讓我們期待著消息中間件可以支持消息回溯功能。

• 重復消費

這個可以交由使用者自己做冪等處理

– 消息需要有序消費

我們知道 Kafka 是分區內消息有序的。當然，需要有序的消息就只能使用一個分區，無疑是以 Kafka 的水平擴展能力作為代價的。如果是需要全局有序，而我們又確定使用 Kafka，而且單分區的吞吐量不能滿足要求，那么我們只能自己進行額外設計來保證了。

2.1 acks配置對消息丟失的影響

2.1.1 acks=1

消息成功寫入 leader 后，就會告訴生產者投遞成功。

如上圖例子，一共三個分區，其中 follower1 和 follower2 均屬于 ISR。假設 leader 成功寫入 3 和 4 之后，ISR 還沒同步，leader 就宕機了，這樣就丟失了 3 和 4，如下圖：

2.1.2 acks=-1 或者 acks=all

消息不僅要成功寫入 leader，還要 ISR 中的所有 follower 同步完成后，才會告知生產者投遞成功。

還是 2.1.1 的例子，這里無非會有兩種情況：

• leader 在同步完成后宕機

• leader 在同步完成前宕機

這個配置對 Kafka 的性能會有較大影響，需要自己斟酌得失。

2.2 unclean.leader.election.enable

這個配置是用來控制 Kafka 是否可以選舉非 ISR 中的副本為 leader，0.11.0.0 之后的版本默認為 false。雖然設置為 true 可以提高 Kafka 的可用性，但是會降低 Kafka 數據的可靠性。

2.3 小結

上面提出的問題均有指出和 Kafka 相關部分的設計是哪些，這里再總結一下：

• 如何避免腦裂問題——了解 Kafka 的 leader 選舉機制
• 數據同步&數據一致性問題——了解 Kafka 的多副本設計
• 消息順序消費問題——了解 Kafka 的日志同步機制（分區有序）

3.副本設計

副本（ Replica ）是分布式系統中常見的概念之一，指的是分布式系統對數據和服務提供的一種冗余方式。

我們知道 Kafka 通過多副本機制，增強了容災備份的能力，并且實現了故障轉移。這無疑是大大提高了 Kafka 的可用性，下面筆者會帶著大家一起看 Kafka 的副本機制是如何設計的。

在此之前，先簡單復習幾個相關的概念：

• 副本是相對分區而言的，即副本是指某個分區的副本
• 在多副本情況下，其中一個副本為 leader，其它均為 follower。只有 leader 對外提供服務，follower 僅同步leader 數據
• 分區中所有的副本集合稱為 AR，ISR 是與 leader 保持同步狀態的副本集合，leader 也是 ISR 中的一員
• LEO 是每個分區下一條消息要寫入的位置
• HW 是 ISR 中最小的 LEO，消費者只能拉取 HW 之前的消息

3.1 失效副本

正常情況下，分區中所有副本都應該屬于 ISR，但是網絡具有不可靠性。因此，難免在某一個時刻會有一些成員會被踢出 ISR，這些副本要么處于同步失效狀態，要么功能失效，這些副本統稱為失效副本。

功能失效指的是無法工作，比如某個 broker 宕機了，那么在它上面的分區就會失效。

同步失效又是怎么判斷是否同步失效的呢？是通過參數 replica.lag.time.max.ms 來判斷的，默認是 10000 毫秒。當某個 follower 同步滯后 leader 的時間超過 10 秒，則判定為同步失效。

具體實現原理如下：

當 follower 將 leader LEO 之前的消息全部同步完成，那么會認為該 follower 已經追上 leader，并更新 lastCaughtUpTimeMs。Kafka 的副本管理器有一個副本過期檢測的定時任務，如果發現當前時間 - lastCaughtUpTimeMs > 10秒，則判定同步失效。

除了時間設置以外，還有另一個參數 replica.lag.max.message（默認4000，這個是 broker 級別的參數），也是用來判定失效副本的。

一般情況下，這兩個參數都是使用默認值就可以，因為如果沒有調優經驗，自己亂配置，容易導致 ISR 變動過于頻繁。同時，需要監控失效副本的數量，因為它是衡量 Kafka 是否健康的一個很重要的指標。

PS：新加入的副本因子/宕機恢復后重新加入的副本在追趕上 leader 之前，也會一直處于失效狀態。

3.1.1 失效副本的作用

失效副本為 Kafka 帶來了什么收益呢？為什么需要設計這么一個狀態呢？

大家不妨試想下：假設允許 ISR 中有一個副本同步一直跟不上 leader。當 leader 發生宕機時，這個 follower 被選舉成了新的 leader，那么這時就會發生消息丟失。

一般會造成副本失效基本是以下兩個原因：

• follower 副本進程卡頓，在一段時間內無法發起同步請求，比如說頻繁發生 FULL GC
• follower 同步過慢，在一段時間內無法追上 leader，比如 I/O有問題（筆者實際工作中遇到過一次，公司搭建自己的物理機房，用了二手服務器，有一臺服務器I/O老化導致讀寫數據慢，導致了副本失效，消息堆積等問題）

3.2 LEO 與 HW

這一小節會更進一步去講解它們之間的關系，讓大家可以更清楚 Kafka 的副本同步機制。

假設現在有 3 個 broker，某個 topic 有 1 個分區，3 個副本。現在有一個 producer 發送了一條消息，這 3 個副本會發生些什么操作。

具體步驟如下：

1. producer 發送消息到 leader
2. leader 將消息追加到日志，并且更新日志的偏移量
3. follower 執行定時任務向 leader 發送 fetch request 同步數據，該請求會帶上自己的 LEO
4. leader 讀取本地日志，并更新 follower 的信息
5. leader 返回 fetch response 給 follower，response 會包含 HW
6. follower 將消息追加到本地日志，并更新日志的偏移量

為了更直觀地理解上面的步驟，下面將會用圖來展示。

1.一個新建的 topic 被寫入了 5 條消息，兩個 follower 去拉取數據

2.leader 給 follower 返回 fetch response，并且 leader 又被寫入了 5 條消息

 其中 follower1 同步了 2 條數據，而 follower2 同步了 3 條數據。

 而 follower 的 HW = min(自己的LEO, 同步回來的HW)

3.follower 再次同步數據，同時 leader 又被寫入了 5 條消息

 leader 更新了 HW

4.leader 給 follower 返回 fetch response

 根據公式，follower 更新 HW = 3

在一個分區中，leader 所在 broker 會記錄所有副本的 LEO 和 自己的 HW；而 follower 所在的 broker 只會記錄自己的 LEO 和 HW。因此，在邏輯層面上，我們可以得到下圖：

0.11.0.0版本之前，Kafka 是基于 HW 的同步機制，但是這個設計有可能出現數據丟失和數據不一致的問題。Kafka 后面的版本通過 leader epoch 來進行優化。

3.3 數據丟失 & 數據不一致的解決方案

3.2小節說到了 LEO 與 HW 的更新機制，并且提到這種設計可能會出現數據丟失和數據不一致。我們先一起來看下這兩個問題是如何產生的。

3.3.1 數據丟失

假設某一分區在某一時刻的狀態如下圖（L 代表是 leader）：

可以看見副本A的 LEO 是 2，HW 是 1；副本B的 LEO 是 2，HW 是 2。顯然，哪怕沒有新的消息寫入副本B中，副本A也要過一小段時間才能追上副本A，并更新 HW。

假設在副本A更新 HW = 2之前，A宕機了，隨后立馬就恢復。這里會有一個截斷機制——根據宕機之前持久化的HW 恢復消息。也就是說，A只恢復了 m1，m2 丟失了。

再假設 A 剛恢復，B 也宕機了，A 成為了 leader。這時 B 又恢復了，并成為了 follower。由于 follower 的 HW 不能比 leader 的 HW 高，所以 B 的 m2 也丟失了。

總結：這里大家可以發現 follower 的 HW 更新是有一定間隙的，像我這個例子其實 follower 是拿到 m2 了，只不過 HW 要追上 leader 需要等下一次的 fetch request。除非配置 acks=-1 并且配置min.insync.replicas 大于 1，unclean.leader.election.enable = true 才行。

3.3.2 數據不一致

假設某一分區在某一時刻，副本A 的 HW = 2，LEO = 2；副本B 的 HW = 1，LEO = 1。

又假設它們同時掛了，B 先恢復。這時，B 會成為 leader，如下圖：

此時，B 寫入新消息 m3，并將 HW、LEO 更新為 2。此時，A 也恢復了。由于 A 的 HW 也是 2，所以沒有截斷消息。如下圖：

這樣一來，A 中 offset = 1 的消息是 m2，B 中 offset = 1 的消息是 m3，數據不一致了。

3.3.3 leader epoch

為了解決 3.3.1 和 3.3.2 的問題，Kafka 從 0.11.0.0 開始引入 leader epoch，在需要截斷時使用 leader epoch 作為依據，而不再是 HW。

如果看框架代碼比較多的同學應該知道 epoch 是相當于版本的這么一個概念。leader epoch 的初始值是 0，每變更一次 leader，leader epoch 就會增加 1。另外，每個副本中還會增加一個矢量<LeaderEpoch => StartOffset>，其中 StartOffset 是當前 leader epoch 下寫入第一條消息的偏移量。每個副本的 Log 下都有一個 leader-epoch-checkpoint 文件，在發生 leader 變更時，會將對應的矢量追加到這個文件中。

3.3.3.1 解決數據丟失問題

還是3.3.1的例子，只不過多了 leader epoch 矢量信息。

副本A：HW=1，LEO=2，LE(leader epoch)=0，Offset(StartOffset)=0

leader-副本B：HW=2，LEO=2，LE=0，Offset(StartOffset)=0

假設在副本A更新 HW = 2之前，A宕機了，隨后立馬就恢復。不過這里不會立馬進行截斷日志操作，而是會發送一個 OffsetsForLeaderEpochRequest 請求給 B，B 作為目前的 leader 在收到請求之后會返回 OffsetsForLeaderEpochResponse 給 A。

我們先來看下 OffsetsForLeaderEpochRequest 和 OffsetsForLeaderEpochResponse 的數據結構。如下圖：

• OffsetsForLeaderEpochRequest

 A 會將自己的 leader epoch 信息給 leader（A的 leader epoch 這里簡化成 LE_A）。這里會出現兩種情況：

– 變更了 leader

 B 會返回 LE_A+1 的 StartOffset 給 A

– 沒有變更 leader

 B 會返回 A 的 LEO 給 A

 因此，我們可以把 OffsetsForLeaderEpochRequest 看作是一個查詢 follower 當前 leader_epoch 的 LEO。

• OffsetsForLeaderEpochResponse

這個例子中，B 會返回2給 A，而此時的 A 的 LEO 剛好是 2，所以不用進行截斷日志。如下圖：

如果此時B掛了，A成了 leader，并有 m3 寫入，就會得到下圖：

可以看見 m2 并沒有丟失，并且也更新了 leader_epoch 矢量為 (1,2)。

3.3.3.2 解決數據不一致問題

上圖是3.3.2的例子。副本A是 leader，B 是 follower。

A 的 HW=2，LEO=2，LE=（0，0）

B 的 HW=1，LEO=1，LE=（0，0）

此時，A 和 B 同時宕機，并且 B 先恢復成為了 leader。此時，epoch 變成了 1。另外，新消息 m3 成功寫入，就會得到下圖：

接著，A 也恢復了，這時 A 不會急著截斷日志，而是給 leader 發送 OffsetsForLeaderEpochRequest，B 會返回 LEO = 1 給 A。因此，A 會截斷日志，刪除 m2。之后，再給 B 發送 fetch request，得到 B 的響應并更新后，將得到下圖：

這樣數據不一致問題就解決了。

這里大家可能會有疑問，m2不是丟失了嗎？是的，這種設計因為更新具有一定的間隙，并且沒有事務管理，所以會有丟失消息的風險。

從 CAP 定理來看，這里的設計屬于 AP。為什么這么說呢？大家不妨想一下，如果為了不丟失數據，這里加了事務控制的設計，那么對于分區而言它的吞吐量是會下降的，甚至是不可用的，因為響應速度是由短板的副本所決定的。對于定位是高吞吐量的 Kafka 而言，這顯然是不可接受的。

3.4 小結

Kafka 通過多副本機制增強了容災備份的能力，并且基于多副本機制實現了故障轉移，避免了單點問題，但同時也引進了新的問題——數據丟失和數據不一致。從 0.11.0.0 版本開始，Kafka 增加了 leader epoch，它對這兩個問題進行了優化。雖然無法完全避免消息丟失，但是從實際的使用角度而言，這個問題其實并不大。有實際工作經驗的同學應該都知道，我們發送消息難以避免需要重推，哪怕消息中間件做到了百分百不丟失，其實我們在使用時仍然會做防止消息丟失的設計。相對而言，數據一致性就更重要了，否則很容易讓訂閱消息的下游系統出現臟數據。

4 leader 選舉機制

在 Kafka 集群中會有一個或者多個 broker，其中有一個 broker 會被選舉為控制器，它負責管理整個集群中所有分區和副本的狀態。分區的 leader 出現故障時，由控制器負責為其選舉新的 leader；當某個分區的 ISR 發生變化時，由控制器負責通知所有 broker 更新其元數據信息；當某個 topic 的分區數量發生變化時，還是由控制器負責分區的重新分配。因此，只要控制器正常工作，分區的 leader 就是唯一的，不會有腦裂問題。

那么， Kafka 是如何保證控制器只有一個的呢？如果控制器發生異常了怎么辦？控制器的選舉和異常恢復又是怎樣的？

4.1 控制器

控制器是 broker 維度的角色，它負責管理整個集群中所有分區和副本的狀態。

Kafka 中的控制器選舉工作依賴于 ZooKeeper，成功競選為控制器的 broker 會在 ZooKeeper 中創建 /controller 臨時節點，節點會存儲以下信息：

{
 "version ": 1,
 "brokerid": 0,
 "timestamp": "1529210278988"
}

其中 version 目前是固定值不用管，brokerid 是成為控制器的 broker 的 id，timestamp 是 broker.id=0 的 broker 成為控制器的時間戳。

在任意時刻，Kafka 集群中有且僅有一個控制器。每個 broker 啟動時會嘗試讀取 ZooKeeper 的 /controller 節點的 brokerid 的值，如果 brokerid ≠ -1，則表示當前集群已有控制器，broker 就會放棄競選；如果不存在 /controller 節點，broker 就會嘗試創建節點，創建成功的 broker 就會成為控制器，將自己的 ID 賦予 brokerid，而對于創建節點失敗的 broker 則會在內存中保存當前控制器的 brokerid 值，這個值標識為 activeControllerId。

上面是啟動 Kafka 集群以及正常情況下添加 broker 情況下的選舉過程。那么當控制器出現故障時，就需要重新選舉了。ZooKeeper 中還有一個與控制器有關的 /controller_epoch 節點，該節點是持久節點，里面存儲了一個整型的 controller_epoch 值，初始值是 1。當控制器發生變化時，controller_epoch 就會加 1。每個和控制器交互的請求一定會帶上 controller_epoch，當控制器發現請求帶上的 controller_epoch 比自己內存的小，那么這個請求則是無效請求；如果請求帶上的 controller_epoch 比自己內存的大，說明自己不再是控制器。由此可見，Kafka 是通過 controller_epoch 來保證控制器的唯一性，進而保證相關操作的一致性。

這里再擴展一下，說下作為控制器的 broker 多出來的責任：

• 監聽分區相關的變化

– 在 ZooKeeper 的 /admin/reassign_partitions 節點注冊 PartitionReassignmentHandler，用來處理分區重分配的動作。

– 在 ZooKeeper 的 /isr_change_notification 節點注冊 IsrChangeNotificetionHandler，用來處理 ISR 集合變更的動作。

– 在 ZooKeeper 的 /admin/preferred-replica-election 節點注冊 PreferredReplicaElectionHandler，用來處理優先副本的選舉動作。

• 監聽主題相關的變化

– 在 ZooKeeper 的 /brokers/topics 節點注冊 TopicChangeHandler，用來處理主題增減的變化。

– 在 ZooKeeper 的 /admin/delete_topics 節點注冊 TopicDeletionHandler，用來處理刪除主題的動作。

• 監聽 broker 相關的變化

 在 ZooKeeper 的 /brokers/ids 節點注冊 BrokerChangeHandler，用來處理 broker 增減的變化。

• 從 ZooKeeper 中讀取當前所有與主題、分區及 broker 有關的信息并進行相應的管理

 對所有主題對應的在 ZooKeeper 中的 /brokers/topics/ 節點添加 PartitionModificationsHandler，用來監聽主題中的分區分配變化。

• 啟動并管理分區狀態機和副本狀態機
• 更新集群的元數據信息
• 如果設置了 auto.leader.rebalance.enable = true，則還會開啟一個名為“auto-leader-rebalance-task”的定時任務來負責維護分區的優先副本的均衡

成功競選控制器的 broker 會在當選后，讀取包括不限于上面提到的在 ZooKeeper 中的節點的數據，初始化上下文信息，并且進行管理。在 Kafka 中，因為會有大量需要讀取或者更新上下文信息的操作，所以這里會有多線程問題。如果單純采用鎖機制實現，那么整體性能會大打折扣。因此，控制器采用的是單線程基于事件隊列的模型。將所有相關的操作、事件進一步封裝成一個個事件，按照事件發生的順序存入 LinkedBlockingQueue 中，最后再使用一個專用線程按 FIFO 的原則處理各個事件。

控制器、非控制器 broker、ZooKeeper 的關系圖如下：

只有控制器會注冊相應的監聽器關注節點的數據變化，其他 broker 則不關注這些節點的數據變化（除了 /controller）。因為所有 broker 都會關心當前的控制器到底是誰，當 /controller 的數據發生變化時，就要更新自己內存中的 activeControllerId。如果原來是控制器的 broker，發現自己現在不是了，就需要關閉資源，如注銷只有控制器才需要的監聽器等。不管什么原因造成 /controller 的 brokerid 變更，再重新選舉控制器之前，要先確定參選的 broker 里面是否有前控制器，如果有，就要先“退位”，再開始新的選舉。

優點：只有控制器注冊監聽器，可以有效避免嚴重依賴 ZooKeeper 的設計的通病——腦裂、羊群效應、ZooKeeper 過載。

5 日志同步機制

在多副本的設計中，要實現數據一致性和順序性，最簡單有效的辦法就是選舉 leader，由 leader 負責寫入順序，follower 復制同步即可。只要 leader 不出問題，如宕機、腦裂等，那么就不需要擔心 follower 的數據同步問題。

不過，一個分布式系統肯定需要考慮故障轉移的。這時就需要考慮 leader 宕機后，選舉新 leader 的問題。上面講了控制器會保證分區 leader 的唯一性，但是數據丟失的問題，還是需要 follower 里面有跟上 leader 的才行。因此，Kafka 里面會有 ISR 這么一個概念。另外，如果需要做到告知客戶端成功提交了某條消息，就需要保證新 leader 里面必需有這條消息，那么需要配置 acks=all 等相關配置。

作者介紹

蔡柱梁，51CTO社區編輯，從事Java后端開發8年，做過傳統項目廣電BOSS系統，后投身互聯網電商，負責過訂單，TMS，中間件等。