聊聊 Kafka： Consumer 源碼解析之 Rebalance 機制

作者：RiemannChow 2021-12-30 22:50:32

Rebalance 本質(zhì)上是一種協(xié)議，規(guī)定了一個 Consumer Group 下的所有 Consumer 如何達成一致，來分配訂閱 Topic 的每個分區(qū)。

[[443148]]

一、前言

我們上一篇分析了 Consumer 如何加入 Consumer Group，其實上一篇是一個很宏觀的東西，主要講 ConsumerCoordinator 怎么與 GroupCoordinator 通信。等等，老周，ConsumerCoordinator 和 GroupCoordinator 是個啥玩意?這兩個組件分別是 Consumer、Kafka Broker 的協(xié)調(diào)器，說白了就是我們設(shè)計模式中的門面模式，具體的內(nèi)容可以看上一篇回顧下。今天這一篇主要講上一篇 Consumer 如何加入 Consumer Group 中的 Rebalance 機制，其實上一篇講了大概了，這一篇更深入的來說一說 Rebalance 機制的具體細節(jié)。

如果你是一個有一定經(jīng)驗的程序員，Rebalance 機制我覺得可以作為一道面試題來考察，而且還是有一定難度的。但是也不需要妄自菲薄，跟著老周的這篇文章下來，相信你一定可以拿下它的。

但有些讀者確實覺得還是有一定難度，別著急，先看下下面 Kafka 的拓撲結(jié)構(gòu)，這個結(jié)構(gòu)很清晰了吧，如果你對 Kafka 的拓撲結(jié)構(gòu)還不了解，那我建議你先別往下看了，先把 Kafka 的拓撲結(jié)構(gòu)搞清楚，或者先看老周前面的幾篇文章再來繼續(xù)閱讀，我覺得效果會更好。

這一篇主要從以下幾點來聊一聊 Rebalance 機制：

什么是 Rebalance 機制?
觸發(fā) Rebalance 機制的時機
Group 狀態(tài)變更
舊版消費者客戶端的問題
Rebalance 機制的原理
Broker 端重平衡場景

二、什么是 Rebalance 機制?

Rebalance 本質(zhì)上是一種協(xié)議，規(guī)定了一個 Consumer Group 下的所有 Consumer 如何達成一致，來分配訂閱 Topic 的每個分區(qū)。

當(dāng)集群中有新成員加入，或者某些主題增加了分區(qū)之后，消費者是怎么進行重新分配消費的?這里就涉及到重平衡(Rebalance)的概念，下面我就給大家講解一下什么是 Kafka 重平衡機制。

從圖中可以找到消費組模型的幾個概念：

同一個消費組，一個分區(qū)只能被一個消費者訂閱消費，但一個消費者可訂閱多個分區(qū)，也就是說每條消息只會被同一個消費組的某一個消費者消費，確保不會被重復(fù)消費;
一個分區(qū)可被不同消費組訂閱，這里有種特殊情況，假如每個消費組只有一個消費者，這樣分區(qū)就會廣播到所有消費者上，實現(xiàn)廣播模式消費。

要想實現(xiàn)以上消費組模型，那么就要實現(xiàn)當(dāng)外部環(huán)境變化時，比如主題新增了分區(qū)，消費組有新成員加入等情況，實現(xiàn)動態(tài)調(diào)整以維持以上模型，那么這個工作就會交給 Kafka 重平衡(Rebalance)機制去處理。

從圖中可看出，Kafka 重平衡是外部觸發(fā)導(dǎo)致的，下面來看下觸發(fā) Kafka 重平衡的時機有哪些。

三、觸發(fā) Rebalance 機制的時機

有新的 Consumer 加入 Consumer Group
有 Consumer 宕機下線。Consumer 并不一定需要真正下線，例如遇到長時間的 GC、網(wǎng)絡(luò)延遲導(dǎo)致消費者長時間未向 GroupCoordinator 發(fā)送 HeartbeatRequest 時，GroupCoordinator 會認為 Consumer 下線。
有 Consumer 主動退出 Consumer Group(發(fā)送 LeaveGroupRequest 請求)。比如客戶端調(diào)用了 unsubscribe() 方法取消對某些主題的訂閱。
Consumer 消費超時，沒有在指定時間內(nèi)提交 offset 偏移量。
Consumer Group 所對應(yīng)的 GroupCoordinator 節(jié)點發(fā)生了變更。
Consumer Group 所訂閱的任一主題或者主題的分區(qū)數(shù)量發(fā)生變化。

四、Group 狀態(tài)變更

4.1 消費端

在 Consumer 側(cè)的門面 ConsumerCoordinator，它繼承了 AbstractCoordinator 抽象類。在協(xié)調(diào)器 AbstractCoordinator 中的內(nèi)部類 MemberState 中我們可以看到協(xié)調(diào)器的四種狀態(tài)，分別是未注冊、重分配后沒收到響應(yīng)、重分配后收到響應(yīng)但還沒有收到分配、穩(wěn)定狀態(tài)。

上述消費端的四種狀態(tài)的轉(zhuǎn)換如下圖所示：

4.2 服務(wù)端

對于 Kafka 服務(wù)端的 GroupCoordinator 則有五種狀態(tài) Empty、PreparingRebalance、CompletingRebalance、Stable、Dead。他們的狀態(tài)轉(zhuǎn)換如下圖所示：

一個消費者組最開始是 Empty
重平衡開啟后，會置于 PreparingRebalance 等待成員加入。
之后變更到 CompletingRebalance 等待分配方案
最后流轉(zhuǎn)到 Stable 完成 Rebalance
當(dāng)有成員變動時，消費者組狀態(tài)從 Stable 變?yōu)?PreparingRebalance。
- 此時所有現(xiàn)存成員需要重新申請加入組
- 當(dāng)所有組成員都退出組后，消費者組狀態(tài)為 Empty。
消費者組處于 Empty 狀態(tài)，Kafka 會定期自動刪除過期 offset。

五、舊版消費者客戶端的問題

ConsumerCoordinator 與 GroupCoordinator 的概念是針對 Kafka 0.9.0 版本后的消費者客戶端而言的，我們暫且把 Kafka 0.9.0 版本之前的消費者客戶端稱為舊版消費者客戶端。舊版消費者客戶端是使用 Zookeeper 的監(jiān)聽器(Watcher)來實現(xiàn)這些功能的。

每個消費組在 Zookeeper 中維護了一個 /consumers//ids 路徑，在此路徑下使用臨時節(jié)點記錄隸屬于此消費組的消費者的唯一標(biāo)識 consumerldString , consumerldString 由消費者啟動時創(chuàng)建。消費者的唯一標(biāo)識由 consumer.id+主機名+時間戳+UUID的部分信息構(gòu)成，其中 consumer.id 是舊版消費者客戶端中的配置，相當(dāng)于新版客戶端中的 client.id。比如某個消費者的唯一標(biāo)識為 consumerld_localhost-1510734527562-64b377f5，那么其中 consumerld 為指定的 consumer.id, localhost 為計算機的主機名，1510734527562代表時間戳，而 64b377f5 表示 UUID 的部分信息。

下圖與 /consumers//ids 同級的還有兩個節(jié)點：owners 和 offsets

/consumers//owners 路徑下記錄了分區(qū)和消費者的對應(yīng)關(guān)系
/consumers//offsets 路徑下記錄了此消費組在分區(qū)中對應(yīng)的消費位移

每個 broker、主題和分區(qū)在 Zookeeper 中也都對應(yīng)一個路徑：

/brokers/ids/記錄了 host、port 及分配在此 broker 上的主題分區(qū)列表;
/brokers/topics/ 記錄了每個分區(qū)的 leader 副本、ISR 集合等信息。
/brokers/topics//partitions//state 記錄了當(dāng)前 leader 副本、leader epoch 等信息。

每個消費者在啟動時都會在 /consumers//ids 和 /brokers/ids 路徑上注冊一個監(jiān)聽器。當(dāng) /consumers//ids 路徑下的子節(jié)點發(fā)生變化時，表示消費組中的消費者發(fā)生了變化;當(dāng) /brokers/ids 路徑下的子節(jié)點發(fā)生變化時，表示 broker 出現(xiàn)了增減。這樣通過 Zookeeper 所提供的 Watcher，每個消費者就可以監(jiān)聽消費組和 Kafka 集群的狀態(tài)了。

這種方式下每個消費者對 Zookeeper 的相關(guān)路徑分別進行監(jiān)聽，當(dāng)觸發(fā)再均衡操作時，一個消費組下的所有消費者會同時進行再均衡操作，而消費者之間并不知道彼此操作的結(jié)果，這樣可能導(dǎo)致 Kafka 工作在一個不正確的狀態(tài)。與此同時，這種嚴(yán)重依賴于 Zookeeper 集群的做法還有兩個比較嚴(yán)重的問題。

羊群效應(yīng)(Herd Effect)：所謂的羊群效應(yīng)是指 Zookeeper 中一個被監(jiān)聽的節(jié)點變化，大量的 Watcher 通知被發(fā)送到客戶端，導(dǎo)致在通知期間的其他操作延遲，也有可能發(fā)生類似死鎖的情況。
腦裂問題(Split Brain)：消費者進行再均衡操作時每個消費者都與 Zookeeper 進行通信以判斷消費者或 broker 變化的情況，由于 Zookeeper 本身的特性，可能導(dǎo)致在同一時刻各個消費者獲取的狀態(tài)不一致，這樣會導(dǎo)致異常問題發(fā)生。

六、Rebalance 機制的原理

Kafka 0.9.0 版本后的消費者客戶端對此進行了重新設(shè)計，將全部消費組分成多個子集，每個消費組的子集在服務(wù)端對應(yīng)一個 GroupCoordinator 對其進行管理，GroupCoordinator 是 Kafka 服務(wù)端中用于管理消費組的組件。而消費者客戶端中的 ConsumerCoordinator 組件負責(zé)與 GroupCoordinator 進行交互。

Rebalance 完整流程需要 Consumer & Coordinator 共同完成
Consumer 端 Rebalance 步驟
- 加入組：對應(yīng) JoinGroup 請求
- 等待 Leader Consumer 分配方案：對應(yīng) SyncGroup 請求
當(dāng)組內(nèi)成員加入組時，Consumer 向協(xié)調(diào)者發(fā)送 JoinGroup 請求。
每個 Consumer 會上報自己訂閱的 topic
Coordinator 收集到所有 JoinGroup 請求后，從這些成員中選擇一個擔(dān)任消費者組的 Leader
- 通常第一個發(fā)送 JoinGroup 請求的自動成為 Leader
Leader Consumer 的任務(wù)是收集所有成員的 topic，根據(jù)信息制定具體的 partition consumer 分配方案。
選出 Leader 后，協(xié)調(diào)者把所有 topic 信息封裝到 JoinGroup Response 中，發(fā)送給 Leader。
Leader Consumer 做出統(tǒng)一分配方案，進入到 SyncGroup 請求。
Leader Consumer 向協(xié)調(diào)者發(fā)送 SyncGroup，將分配方案發(fā)給協(xié)調(diào)者。
其他成員也會發(fā)出 SyncGroup 請求
協(xié)調(diào)者以 SyncGroup Response 的方式將方案下發(fā)給所有成員