你能說(shuō)出Kafka這些原理嗎

作者：cxuan 2019-12-30 09:28:53

開(kāi)源

Kafka 如果你只追一個(gè)妹子并且對(duì)這個(gè)妹子特別用心的話，知道的人一定會(huì)說(shuō)你是個(gè)好男人;如果你只是淺嘗輒止并且追了大部分妹子的話，知道的人一定會(huì)罵你渣男。

作為一家“創(chuàng)意+整合+營(yíng)銷(xiāo)”的成都網(wǎng)站建設(shè)機(jī)構(gòu)，我們?cè)跇I(yè)內(nèi)良好的客戶口碑。成都創(chuàng)新互聯(lián)提供從前期的網(wǎng)站品牌分析策劃、網(wǎng)站設(shè)計(jì)、網(wǎng)站設(shè)計(jì)制作、成都網(wǎng)站制作、創(chuàng)意表現(xiàn)、網(wǎng)頁(yè)制作、系統(tǒng)開(kāi)發(fā)以及后續(xù)網(wǎng)站營(yíng)銷(xiāo)運(yùn)營(yíng)等一系列服務(wù)，幫助企業(yè)打造創(chuàng)新的互聯(lián)網(wǎng)品牌經(jīng)營(yíng)模式與有效的網(wǎng)絡(luò)營(yíng)銷(xiāo)方法,創(chuàng)造更大的價(jià)值。

如果你只追一個(gè)妹子并且對(duì)這個(gè)妹子特別用心的話，知道的人一定會(huì)說(shuō)你是個(gè)好男人;如果你只是淺嘗輒止并且追了大部分妹子的話，知道的人一定會(huì)罵你渣男。

做技術(shù)也是一樣的，如果你對(duì)一門(mén)技術(shù)鉆研的特別深的話，那你一定是這個(gè)領(lǐng)域不可或缺的人才;如果你每個(gè)技術(shù)都想學(xué)并且遇到一些困難就退縮，那么你就離被替代不遠(yuǎn)了。

中國(guó)現(xiàn)在的社會(huì)就像是一劑催化劑，催生的都是快節(jié)奏的人。周末我聽(tīng)到了這么一個(gè)事情：在我的大學(xué)里，有這樣一個(gè)人，他是班長(zhǎng)。家里有些錢(qián)，可能接觸社會(huì)比較早，為人處事比較成熟，和導(dǎo)員關(guān)系非常好，可以說(shuō)好到經(jīng)常性的不來(lái)上課，導(dǎo)員竟然還幫忙寫(xiě)假條;好到同學(xué)評(píng)價(jià)班長(zhǎng)的時(shí)候，打分60分導(dǎo)員竟然私自幫忙改到80分。我說(shuō)了一句話：這是社會(huì)催化的產(chǎn)物，校園也是一個(gè)小社會(huì)，很多事情從這個(gè)小社會(huì)就開(kāi)始變質(zhì)了。

我們從來(lái)不思考自己得到的是不是合規(guī)合法的。我記得我的一個(gè)人生導(dǎo)師和我說(shuō)過(guò)這樣一句話：好好做人，別總是想尋求捷徑。這句話像茶一樣，慢慢品才香。

下面回到技術(shù)。

如果只是為了開(kāi)發(fā) Kafka 應(yīng)用程序，或者只是在生產(chǎn)環(huán)境使用 Kafka，那么了解 Kafka 的內(nèi)部工作原理不是必須的。不過(guò)，了解 Kafka 的內(nèi)部工作原理有助于理解 Kafka 的行為，也利用快速診斷問(wèn)題。下面我們來(lái)探討一下這三個(gè)問(wèn)題

• Kafka 是如何進(jìn)行復(fù)制的
• Kafka 是如何處理來(lái)自生產(chǎn)者和消費(fèi)者的請(qǐng)求的
• Kafka 的存儲(chǔ)細(xì)節(jié)是怎樣的

如果感興趣的話，就請(qǐng)花費(fèi)你一些時(shí)間，耐心看完這篇文章。

集群成員間的關(guān)系

我們知道，Kafka 是運(yùn)行在 ZooKeeper 之上的，因?yàn)?ZooKeeper 是以集群形式出現(xiàn)的，所以 Kafka 也可以以集群形式出現(xiàn)。這也就涉及到多個(gè)生產(chǎn)者和多個(gè)消費(fèi)者如何協(xié)調(diào)的問(wèn)題，這個(gè)維護(hù)集群間的關(guān)系也是由 ZooKeeper 來(lái)完成的。你應(yīng)該會(huì)知道，Kafka 集群間會(huì)有多個(gè) 主機(jī)(broker)，每個(gè) broker 都會(huì)有一個(gè) broker.id，每個(gè) broker.id 都有一個(gè)唯一的標(biāo)識(shí)符用來(lái)區(qū)分，這個(gè)標(biāo)識(shí)符可以在配置文件里手動(dòng)指定，也可以自動(dòng)生成。

Kafka 可以通過(guò) broker.id.generation.enable 和 reserved.broker.max.id 來(lái)配合生成新的 broker.id。

broker.id.generation.enable參數(shù)是用來(lái)配置是否開(kāi)啟自動(dòng)生成 broker.id 的功能，默認(rèn)情況下為true，即開(kāi)啟此功能。自動(dòng)生成的broker.id有一個(gè)默認(rèn)值，默認(rèn)值為1000，也就是說(shuō)默認(rèn)情況下自動(dòng)生成的 broker.id 從1001開(kāi)始。

Kafka 在啟動(dòng)時(shí)會(huì)在 ZooKeeper 中 /brokers/ids 路徑下注冊(cè)一個(gè)與當(dāng)前 broker 的 id 相同的臨時(shí)節(jié)點(diǎn)。Kafka 的健康狀態(tài)檢查就依賴于此節(jié)點(diǎn)。當(dāng)有 broker 加入集群或者退出集群時(shí)，這些組件就會(huì)獲得通知。

• 如果你要啟動(dòng)另外一個(gè)具有相同 ID 的 broker，那么就會(huì)得到一個(gè)錯(cuò)誤 —— 新的 broker 會(huì)試著進(jìn)行注冊(cè)，但不會(huì)成功，因?yàn)?ZooKeeper 里面已經(jīng)有一個(gè)相同 ID 的 broker。
• 在 broker 停機(jī)、出現(xiàn)分區(qū)或者長(zhǎng)時(shí)間垃圾回收停頓時(shí)，broker 會(huì)從 ZooKeeper 上斷開(kāi)連接，此時(shí) broker 在啟動(dòng)時(shí)創(chuàng)建的臨時(shí)節(jié)點(diǎn)會(huì)從 ZooKeeper 中移除。監(jiān)聽(tīng) broker 列表的 Kafka 組件會(huì)被告知該 broker 已移除。
• 在關(guān)閉 broker 時(shí)，它對(duì)應(yīng)的節(jié)點(diǎn)也會(huì)消失，不過(guò)它的 ID 會(huì)繼續(xù)存在其他數(shù)據(jù)結(jié)構(gòu)中，例如主題的副本列表中，副本列表復(fù)制我們下面再說(shuō)。在完全關(guān)閉一個(gè) broker 之后，如果使用相同的 ID 啟動(dòng)另一個(gè)全新的 broker，它會(huì)立刻加入集群，并擁有一個(gè)與舊 broker 相同的分區(qū)和主題。

Broker Controller 的作用

我們之前在講 Kafka Rebalance 重平衡的時(shí)候，提過(guò)一個(gè)群組協(xié)調(diào)器，負(fù)責(zé)協(xié)調(diào)群組間的關(guān)系，那么 broker 之間也有一個(gè)控制器組件(Controller)，它是 Kafka 的核心組件。它的主要作用是在 ZooKeeper 的幫助下管理和協(xié)調(diào)整個(gè) Kafka 集群，集群中的每個(gè) broker 都可以稱為 controller，但是在 Kafka 集群?jiǎn)?dòng)后，只有一個(gè) broker 會(huì)成為 Controller 。既然 Kafka 集群是依賴于 ZooKeeper 集群的，在這里就簡(jiǎn)單提一下 znode 節(jié)點(diǎn)的問(wèn)題。

ZooKeeper 的數(shù)據(jù)是保存在節(jié)點(diǎn)上的，每個(gè)節(jié)點(diǎn)也被稱為znode，znode 節(jié)點(diǎn)是一種樹(shù)形的文件結(jié)構(gòu)，它很像 Linux 操作系統(tǒng)的文件路徑，ZooKeeper 的根節(jié)點(diǎn)是 /。

znode 根據(jù)數(shù)據(jù)的持久化方式可分為臨時(shí)節(jié)點(diǎn)和持久性節(jié)點(diǎn)。持久性節(jié)點(diǎn)不會(huì)因?yàn)?ZooKeeper 狀態(tài)的變化而消失，但是臨時(shí)節(jié)點(diǎn)會(huì)隨著 ZooKeeper 的重啟而自動(dòng)消失。

znode 節(jié)點(diǎn)有一個(gè) Watcher 機(jī)制：當(dāng)數(shù)據(jù)發(fā)生變化的時(shí)候， ZooKeeper 會(huì)產(chǎn)生一個(gè) Watcher 事件，并且會(huì)發(fā)送到客戶端。Watcher 監(jiān)聽(tīng)機(jī)制是 Zookeeper 中非常重要的特性，我們基于 Zookeeper 上創(chuàng)建的節(jié)點(diǎn)，可以對(duì)這些節(jié)點(diǎn)綁定監(jiān)聽(tīng)事件，比如可以監(jiān)聽(tīng)節(jié)點(diǎn)數(shù)據(jù)變更、節(jié)點(diǎn)刪除、子節(jié)點(diǎn)狀態(tài)變更等事件，通過(guò)這個(gè)事件機(jī)制，可以基于 ZooKeeper 實(shí)現(xiàn)分布式鎖、集群管理等功能。

控制器的選舉

Kafka 當(dāng)前選舉控制器的規(guī)則是：Kafka 集群中第一個(gè)啟動(dòng)的 broker 通過(guò)在 ZooKeeper 里創(chuàng)建一個(gè)臨時(shí)節(jié)點(diǎn) /controller 讓自己成為 controller 控制器。其他 broker 在啟動(dòng)時(shí)也會(huì)嘗試創(chuàng)建這個(gè)節(jié)點(diǎn)，但是由于這個(gè)節(jié)點(diǎn)已存在，所以后面想要?jiǎng)?chuàng)建 /controller 節(jié)點(diǎn)時(shí)就會(huì)收到一個(gè)節(jié)點(diǎn)已存在 的異常。然后其他 broker 會(huì)在這個(gè)控制器上注冊(cè)一個(gè) ZooKeeper 的 watch 對(duì)象，/controller 節(jié)點(diǎn)發(fā)生變化時(shí)，其他 broker 就會(huì)收到節(jié)點(diǎn)變更通知。這種方式可以確保只有一個(gè)控制器存在。那么只有單獨(dú)的節(jié)點(diǎn)一定是有個(gè)問(wèn)題的，那就是單點(diǎn)問(wèn)題。

如果控制器關(guān)閉或者與 ZooKeeper 斷開(kāi)鏈接，ZooKeeper 上的臨時(shí)節(jié)點(diǎn)就會(huì)消失。集群中的其他節(jié)點(diǎn)收到 watch 對(duì)象發(fā)送控制器下線的消息后，其他 broker 節(jié)點(diǎn)都會(huì)嘗試讓自己去成為新的控制器。其他節(jié)點(diǎn)的創(chuàng)建規(guī)則和第一個(gè)節(jié)點(diǎn)的創(chuàng)建原則一致，都是第一個(gè)在 ZooKeeper 里成功創(chuàng)建控制器節(jié)點(diǎn)的 broker 會(huì)成為新的控制器，那么其他節(jié)點(diǎn)就會(huì)收到節(jié)點(diǎn)已存在的異常，然后在新的控制器節(jié)點(diǎn)上再次創(chuàng)建 watch 對(duì)象進(jìn)行監(jiān)聽(tīng)。

控制器的作用

那么說(shuō)了這么多，控制是什么呢?控制器的作用是什么呢?或者說(shuō)控制器的這么一個(gè)組件被設(shè)計(jì)用來(lái)干什么?別著急，接下來(lái)我們就要說(shuō)一說(shuō)。

Kafka 被設(shè)計(jì)為一種模擬狀態(tài)機(jī)的多線程控制器，它可以作用有下面這幾點(diǎn)

• 控制器相當(dāng)于部門(mén)(集群)中的部門(mén)經(jīng)理(broker controller)，用于管理部門(mén)中的部門(mén)成員(broker)
• 控制器是所有 broker 的一個(gè)監(jiān)視器，用于監(jiān)控 broker 的上線和下線
• 在 broker 宕機(jī)后，控制器能夠選舉新的分區(qū) Leader
• 控制器能夠和 broker 新選取的 Leader 發(fā)送消息

再細(xì)分一下可以具體分為如下 5 點(diǎn)

• 主題管理 : Kafka Controller 可以幫助我們完成對(duì) Kafka 主題創(chuàng)建、刪除和增加分區(qū)的操作，簡(jiǎn)而言之就是對(duì)分區(qū)擁有最高行使權(quán)。

換句話說(shuō)，當(dāng)我們執(zhí)行kafka-topics 腳本時(shí)，大部分的后臺(tái)工作都是控制器來(lái)完成的。

• 分區(qū)重分配: 分區(qū)重分配主要是指，kafka-reassign-partitions 腳本提供的對(duì)已有主題分區(qū)進(jìn)行細(xì)粒度的分配功能。這部分功能也是控制器實(shí)現(xiàn)的。
• Prefered 領(lǐng)導(dǎo)者選舉 : Preferred 領(lǐng)導(dǎo)者選舉主要是 Kafka 為了避免部分 Broker 負(fù)載過(guò)重而提供的一種換 Leader 的方案。
• 集群成員管理: 主要管理 新增 broker、broker 關(guān)閉、broker 宕機(jī)
• 數(shù)據(jù)服務(wù): 控制器的最后一大類工作，就是向其他 broker 提供數(shù)據(jù)服務(wù)。控制器上保存了最全的集群元數(shù)據(jù)信息，其他所有 broker 會(huì)定期接收控制器發(fā)來(lái)的元數(shù)據(jù)更新請(qǐng)求，從而更新其內(nèi)存中的緩存數(shù)據(jù)。這些數(shù)據(jù)我們會(huì)在下面討論

當(dāng)控制器發(fā)現(xiàn)一個(gè) broker 離開(kāi)集群(通過(guò)觀察相關(guān) ZooKeeper 路徑)，控制器會(huì)收到消息：這個(gè) broker 所管理的那些分區(qū)需要一個(gè)新的 Leader?？刂破鲿?huì)依次遍歷每個(gè)分區(qū)，確定誰(shuí)能夠作為新的 Leader，然后向所有包含新 Leader 或現(xiàn)有 Follower 的分區(qū)發(fā)送消息，該請(qǐng)求消息包含誰(shuí)是新的 Leader 以及誰(shuí)是 Follower 的信息。隨后，新的 Leader 開(kāi)始處理來(lái)自生產(chǎn)者和消費(fèi)者的請(qǐng)求，F(xiàn)ollower 用于從新的 Leader 那里進(jìn)行復(fù)制。

這就很像外包公司的一個(gè)部門(mén)，這個(gè)部門(mén)就是專門(mén)出差的，每個(gè)人在不同的地方辦公，但是中央總部有一個(gè)部門(mén)經(jīng)理，現(xiàn)在部門(mén)經(jīng)理突然離職了。公司不打算外聘人員，決定從部門(mén)內(nèi)部選一個(gè)能力強(qiáng)的人當(dāng)領(lǐng)導(dǎo)，然后當(dāng)上領(lǐng)導(dǎo)的人需要向自己的組員發(fā)送消息，這條消息就是任命消息和明確他管理了哪些人，大家都知道了，然后再各自給部門(mén)干活。

當(dāng)控制器發(fā)現(xiàn)一個(gè) broker 加入集群時(shí)，它會(huì)使用 broker ID 來(lái)檢查新加入的 broker 是否包含現(xiàn)有分區(qū)的副本。如果有控制器就會(huì)把消息發(fā)送給新加入的 broker 和 現(xiàn)有的 broker。

上面這塊關(guān)于分區(qū)復(fù)制的內(nèi)容我們接下來(lái)會(huì)說(shuō)到。

broker controller 數(shù)據(jù)存儲(chǔ)

上面我們介紹到 broker controller 會(huì)提供數(shù)據(jù)服務(wù)，用于保存大量的 Kafka 集群數(shù)據(jù)。如下圖

可以對(duì)上面保存信息歸類，主要分為三類：

• broker 上的所有信息，包括 broker 中的所有分區(qū)，broker 所有分區(qū)副本，當(dāng)前都有哪些運(yùn)行中的 broker，哪些正在關(guān)閉中的 broker 。
• 所有主題信息，包括具體的分區(qū)信息，比如領(lǐng)導(dǎo)者副本是誰(shuí)，ISR 集合中有哪些副本等。
• 所有涉及運(yùn)維任務(wù)的分區(qū)。包括當(dāng)前正在進(jìn)行 Preferred 領(lǐng)導(dǎo)者選舉以及分區(qū)重分配的分區(qū)列表。

Kafka 是離不開(kāi) ZooKeeper的，所以這些數(shù)據(jù)信息在 ZooKeeper 中也保存了一份。每當(dāng)控制器初始化時(shí)，它都會(huì)從 ZooKeeper 上讀取對(duì)應(yīng)的元數(shù)據(jù)并填充到自己的緩存中。

broker controller 故障轉(zhuǎn)移

我們?cè)谇懊嬲f(shuō)過(guò)，第一個(gè)在 ZooKeeper 中的 /brokers/ids下創(chuàng)建節(jié)點(diǎn)的 broker 作為 broker controller，也就是說(shuō) broker controller 只有一個(gè)，那么必然會(huì)存在單點(diǎn)失效問(wèn)題。kafka 為考慮到這種情況提供了故障轉(zhuǎn)移功能，也就是 Fail Over。如下圖：

最一開(kāi)始，broker1 會(huì)搶先注冊(cè)成功成為 controller，然后由于網(wǎng)絡(luò)抖動(dòng)或者其他原因致使 broker1 掉線，ZooKeeper 通過(guò) Watch 機(jī)制覺(jué)察到 broker1 的掉線，之后所有存活的 brokers 開(kāi)始競(jìng)爭(zhēng)成為 controller，這時(shí) broker3 搶先注冊(cè)成功，此時(shí) ZooKeeper 存儲(chǔ)的 controller 信息由 broker1 -> broker3，之后，broker3 會(huì)從 ZooKeeper 中讀取元數(shù)據(jù)信息，并初始化到自己的緩存中。

注意：ZooKeeper 中存儲(chǔ)的不是緩存信息，broker 中存儲(chǔ)的才是緩存信息。

broker controller 存在的問(wèn)題

在 Kafka 0.11 版本之前，控制器的設(shè)計(jì)是相當(dāng)繁瑣的。我們上面提到過(guò)一句話：Kafka controller 被設(shè)計(jì)為一種模擬狀態(tài)機(jī)的多線程控制器，這種設(shè)計(jì)其實(shí)是存在一些問(wèn)題的。

• controller 狀態(tài)的更改由不同的監(jiān)聽(tīng)器并發(fā)執(zhí)行，因此需要進(jìn)行很復(fù)雜的同步，并且容易出錯(cuò)而且難以調(diào)試。
• 狀態(tài)傳播不同步，broker 可能在時(shí)間不確定的情況下出現(xiàn)多種狀態(tài)，這會(huì)導(dǎo)致不必要的額外的數(shù)據(jù)丟失
• controller 控制器還會(huì)為主題刪除創(chuàng)建額外的 I/O 線程，導(dǎo)致性能損耗
• controller 的多線程設(shè)計(jì)還會(huì)訪問(wèn)共享數(shù)據(jù)，我們知道，多線程訪問(wèn)共享數(shù)據(jù)是線程同步最麻煩的地方，為了保護(hù)數(shù)據(jù)安全性，控制器不得不在代碼中大量使用ReentrantLock 同步機(jī)制，這就進(jìn)一步拖慢了整個(gè)控制器的處理速度。

broker controller 內(nèi)部設(shè)計(jì)原理

在 Kafka 0.11 之后，Kafka controller 采用了新的設(shè)計(jì)，把多線程的方案改成了單線程加事件隊(duì)列的方案。如下圖所示：

主要所做的改變有下面這幾點(diǎn)：

第一個(gè)改進(jìn)是增加了一個(gè) Event Executor Thread，事件執(zhí)行線程，從圖中可以看出，不管是 Event Queue 事件隊(duì)列還是 Controller context 控制器上下文都會(huì)交給事件執(zhí)行線程進(jìn)行處理。將原來(lái)執(zhí)行的操作全部建模成一個(gè)個(gè)獨(dú)立的事件，發(fā)送到專屬的事件隊(duì)列中，供此線程消費(fèi)。

第二個(gè)改進(jìn)是將之前同步的 ZooKeeper 全部改為異步操作。ZooKeeper API 提供了兩種讀寫(xiě)的方式：同步和異步。之前控制器操作 ZooKeeper 都是采用的同步方式，這次把同步方式改為異步，據(jù)測(cè)試，效率提升了10倍。

第三個(gè)改進(jìn)是根據(jù)優(yōu)先級(jí)處理請(qǐng)求，之前的設(shè)計(jì)是 broker 會(huì)公平性的處理所有 controller 發(fā)送的請(qǐng)求。什么意思呢?公平性難道還不好嗎?在某些情況下是的，比如 broker 在排隊(duì)處理 produce 請(qǐng)求，這時(shí)候 controller 發(fā)出了一個(gè) StopReplica 的請(qǐng)求，你會(huì)怎么辦?還在繼續(xù)處理 produce 請(qǐng)求嗎?這個(gè) produce 請(qǐng)求還有用嗎?此時(shí)最合理的處理順序應(yīng)該是，賦予 StopReplica 請(qǐng)求更高的優(yōu)先級(jí)，使它能夠得到搶占式的處理。

副本機(jī)制

復(fù)制功能是 Kafka 架構(gòu)的核心功能，在 Kafka 文檔里面 Kafka 把自己描述為 一個(gè)分布式的、可分區(qū)的、可復(fù)制的提交日志服務(wù)。復(fù)制之所以這么關(guān)鍵，是因?yàn)橄⒌某志么鎯?chǔ)非常重要，這能夠保證在主節(jié)點(diǎn)宕機(jī)后依舊能夠保證 Kafka 高可用。副本機(jī)制也可以稱為備份機(jī)制(Replication)，通常指分布式系統(tǒng)在多臺(tái)網(wǎng)絡(luò)交互的機(jī)器上保存有相同的數(shù)據(jù)備份/拷貝。

Kafka 使用主題來(lái)組織數(shù)據(jù)，每個(gè)主題又被分為若干個(gè)分區(qū)，分區(qū)會(huì)部署在一到多個(gè) broker 上，每個(gè)分區(qū)都會(huì)有多個(gè)副本，所以副本也會(huì)被保存在 broker 上，每個(gè) broker 可能會(huì)保存成千上萬(wàn)個(gè)副本。下圖是一個(gè)副本復(fù)制示意圖：

如上圖所示，為了簡(jiǎn)單我只畫(huà)出了兩個(gè) broker ,每個(gè) broker 指保存了一個(gè) Topic 的消息，在 broker1 中分區(qū)0 是Leader，它負(fù)責(zé)進(jìn)行分區(qū)的復(fù)制工作，把 broker1 中的分區(qū)0復(fù)制一個(gè)副本到 broker2 的主題 A 的分區(qū)0。同理，主題 A 的分區(qū)1也是一樣的道理。

副本類型分為兩種：一種是 Leader(領(lǐng)導(dǎo)者) 副本，一種是Follower(跟隨者)副本。

Leader 副本

Kafka 在創(chuàng)建分區(qū)的時(shí)候都要選舉一個(gè)副本，這個(gè)選舉出來(lái)的副本就是 Leader 領(lǐng)導(dǎo)者副本。

Follower 副本

除了 Leader 副本以外的副本統(tǒng)稱為 Follower 副本，F(xiàn)ollower 不對(duì)外提供服務(wù)。下面是 Leader 副本的工作方式。

這幅圖需要注意以下幾點(diǎn)：

• Kafka 中，F(xiàn)ollower 副本也就是追隨者副本是不對(duì)外提供服務(wù)的。這就是說(shuō)，任何一個(gè)追隨者副本都不能響應(yīng)消費(fèi)者和生產(chǎn)者的請(qǐng)求。所有的請(qǐng)求都是由領(lǐng)導(dǎo)者副本來(lái)處理。或者說(shuō)，所有的請(qǐng)求都必須發(fā)送到 Leader 副本所在的 broker 中，F(xiàn)ollower 副本只是用作數(shù)據(jù)拉取，采用異步拉取的方式，并寫(xiě)入到自己的提交日志中，從而實(shí)現(xiàn)與 Leader 的同步
• 當(dāng) Leader 副本所在的 broker 宕機(jī)后，Kafka 依托于 ZooKeeper 提供的監(jiān)控功能能夠?qū)崟r(shí)感知到，并開(kāi)啟新一輪的選舉，從追隨者副本中選一個(gè)作為 Leader。如果宕機(jī)的 broker 重啟完成后，該分區(qū)的副本會(huì)作為 Follower 重新加入。

首領(lǐng)的另一個(gè)任務(wù)是搞清楚哪個(gè)跟隨者的狀態(tài)與自己是一致的。跟隨者為了保證與領(lǐng)導(dǎo)者的狀態(tài)一致，在有新消息到達(dá)之前先嘗試從領(lǐng)導(dǎo)者那里復(fù)制消息。為了與領(lǐng)導(dǎo)者保持一致，跟隨者向領(lǐng)導(dǎo)者發(fā)起獲取數(shù)據(jù)的請(qǐng)求，這種請(qǐng)求與消費(fèi)者為了讀取消息而發(fā)送的信息是一樣的。

跟隨者向領(lǐng)導(dǎo)者發(fā)送消息的過(guò)程是這樣的，先請(qǐng)求消息1，然后再接收到消息1，在時(shí)候到請(qǐng)求1之后，發(fā)送請(qǐng)求2，在收到領(lǐng)導(dǎo)者給發(fā)送給跟隨者之前，跟隨者是不會(huì)繼續(xù)發(fā)送消息的。這個(gè)過(guò)程如下：

跟隨者副本在收到響應(yīng)消息前，是不會(huì)繼續(xù)發(fā)送消息，這一點(diǎn)很重要。通過(guò)查看每個(gè)跟隨者請(qǐng)求的最新偏移量，首領(lǐng)就會(huì)知道每個(gè)跟隨者復(fù)制的進(jìn)度。如果跟隨者在10s 內(nèi)沒(méi)有請(qǐng)求任何消息，或者雖然跟隨者已經(jīng)發(fā)送請(qǐng)求，但是在10s 內(nèi)沒(méi)有收到消息，就會(huì)被認(rèn)為是不同步的。如果一個(gè)副本沒(méi)有與領(lǐng)導(dǎo)者同步，那么在領(lǐng)導(dǎo)者掉線后，這個(gè)副本將不會(huì)稱為領(lǐng)導(dǎo)者，因?yàn)檫@個(gè)副本的消息不是全部的。

與之相反的，如果跟隨者同步的消息和領(lǐng)導(dǎo)者副本的消息一致，那么這個(gè)跟隨者副本又被稱為同步的副本。也就是說(shuō)，如果領(lǐng)導(dǎo)者掉線，那么只有同步的副本能夠稱為領(lǐng)導(dǎo)者。

關(guān)于副本機(jī)制我們說(shuō)了這么多，那么副本機(jī)制的好處是什么呢?

• 能夠立刻看到寫(xiě)入的消息，就是你使用生產(chǎn)者 API 成功向分區(qū)寫(xiě)入消息后，馬上使用消費(fèi)者就能讀取剛才寫(xiě)入的消息
• 能夠?qū)崿F(xiàn)消息的冪等性，啥意思呢?就是對(duì)于生產(chǎn)者產(chǎn)生的消息，在消費(fèi)者進(jìn)行消費(fèi)的時(shí)候，它每次都會(huì)看到消息存在，并不會(huì)存在消息不存在的情況

同步復(fù)制和異步復(fù)制

我在學(xué)習(xí)副本機(jī)制的時(shí)候，有個(gè)疑問(wèn)，既然領(lǐng)導(dǎo)者副本和跟隨者副本是發(fā)送 - 等待機(jī)制的，這是一種同步的復(fù)制方式，那么為什么說(shuō)跟隨者副本同步領(lǐng)導(dǎo)者副本的時(shí)候是一種異步操作呢?

我認(rèn)為是這樣的，跟隨者副本在同步領(lǐng)導(dǎo)者副本后會(huì)把消息保存在本地 log 中，這個(gè)時(shí)候跟隨者會(huì)給領(lǐng)導(dǎo)者副本一個(gè)響應(yīng)消息，告訴領(lǐng)導(dǎo)者自己已經(jīng)保存成功了，同步復(fù)制的領(lǐng)導(dǎo)者會(huì)等待所有的跟隨者副本都寫(xiě)入成功后，再返回給 producer 寫(xiě)入成功的消息。而異步復(fù)制是領(lǐng)導(dǎo)者副本不需要關(guān)心跟隨者副本是否寫(xiě)入成功，只要領(lǐng)導(dǎo)者副本自己把消息保存到本地 log ，就會(huì)返回給 producer 寫(xiě)入成功的消息。下面是同步復(fù)制和異步復(fù)制的過(guò)程：

同步復(fù)制

• producer 通知 ZooKeeper 識(shí)別領(lǐng)導(dǎo)者
• producer 向領(lǐng)導(dǎo)者寫(xiě)入消息
• 領(lǐng)導(dǎo)者收到消息后會(huì)把消息寫(xiě)入到本地 log
• 跟隨者會(huì)從領(lǐng)導(dǎo)者那里拉取消息
• 跟隨者向本地寫(xiě)入 log
• 跟隨者向領(lǐng)導(dǎo)者發(fā)送寫(xiě)入成功的消息
• 領(lǐng)導(dǎo)者會(huì)收到所有的跟隨者發(fā)送的消息
• 領(lǐng)導(dǎo)者向 producer 發(fā)送寫(xiě)入成功的消息

異步復(fù)制

和同步復(fù)制的區(qū)別在于，領(lǐng)導(dǎo)者在寫(xiě)入本地log之后，直接向客戶端發(fā)送寫(xiě)入成功消息，不需要等待所有跟隨者復(fù)制完成。

ISR

Kafka動(dòng)態(tài)維護(hù)了一個(gè)同步狀態(tài)的副本的集合(a set of In-Sync Replicas)，簡(jiǎn)稱ISR，ISR 也是一個(gè)很重要的概念，我們之前說(shuō)過(guò)，追隨者副本不提供服務(wù)，只是定期的異步拉取領(lǐng)導(dǎo)者副本的數(shù)據(jù)而已，拉取這個(gè)操作就相當(dāng)于是復(fù)制，ctrl-c + ctrl-v大家肯定用的熟。那么是不是說(shuō) ISR 集合中的副本消息的數(shù)量都會(huì)與領(lǐng)導(dǎo)者副本消息數(shù)量一樣呢?那也不一定，判斷的依據(jù)是 broker 中參數(shù) replica.lag.time.max.ms 的值，這個(gè)參數(shù)的含義就是跟隨者副本能夠落后領(lǐng)導(dǎo)者副本最長(zhǎng)的時(shí)間間隔。

replica.lag.time.max.ms 參數(shù)默認(rèn)的時(shí)間是 10秒，如果跟隨者副本落后領(lǐng)導(dǎo)者副本的時(shí)間不超過(guò) 10秒，那么 Kafka 就認(rèn)為領(lǐng)導(dǎo)者和跟隨者是同步的。即使此時(shí)跟隨者副本中存儲(chǔ)的消息要小于領(lǐng)導(dǎo)者副本。如果跟隨者副本要落后于領(lǐng)導(dǎo)者副本 10秒以上的話，跟隨者副本就會(huì)從 ISR 被剔除。倘若該副本后面慢慢地追上了領(lǐng)導(dǎo)者的進(jìn)度，那么它是能夠重新被加回 ISR 的。這也表明，ISR 是一個(gè)動(dòng)態(tài)調(diào)整的集合，而非靜態(tài)不變的。

Unclean 領(lǐng)導(dǎo)者選舉

既然 ISR 是可以動(dòng)態(tài)調(diào)整的，那么必然會(huì)出現(xiàn) ISR 集合中為空的情況，由于領(lǐng)導(dǎo)者副本是一定出現(xiàn)在 ISR 集合中的，那么 ISR 集合為空必然說(shuō)明領(lǐng)導(dǎo)者副本也掛了，所以此時(shí) Kafka 需要重新選舉一個(gè)新的領(lǐng)導(dǎo)者，那么該如何選舉呢?現(xiàn)在你需要轉(zhuǎn)變一下思路，我們上面說(shuō) ISR 集合中一定是與領(lǐng)導(dǎo)者同步的副本，那么不再 ISR 集合中的副本一定是不與領(lǐng)導(dǎo)者同步的副本了，也就是不再 ISR 列表中的跟隨者副本會(huì)丟失一些消息。如果你開(kāi)啟 broker 端參數(shù) unclean.leader.election.enable的話，下一個(gè)領(lǐng)導(dǎo)者就會(huì)在這些非同步的副本中選舉。這種選舉也叫做Unclean 領(lǐng)導(dǎo)者選舉。

如果你接觸過(guò)分布式項(xiàng)目的話你一定知道 CAP 理論，那么這種 Unclean 領(lǐng)導(dǎo)者選舉其實(shí)是犧牲了數(shù)據(jù)一致性，保證了 Kafka 的高可用性。

你可以根據(jù)你的實(shí)際業(yè)務(wù)場(chǎng)景決定是否開(kāi)啟 Unclean 領(lǐng)導(dǎo)者選舉，一般不建議開(kāi)啟這個(gè)參數(shù)，因?yàn)閿?shù)據(jù)的一致性要比可用性重要的多。

Kafka 請(qǐng)求處理流程

broker 的大部分工作是處理客戶端、分區(qū)副本和控制器發(fā)送給分區(qū)領(lǐng)導(dǎo)者的請(qǐng)求。這種請(qǐng)求一般都是請(qǐng)求/響應(yīng)式的，我猜測(cè)你接觸最早的請(qǐng)求/響應(yīng)的方式應(yīng)該就是 HTTP 請(qǐng)求了。事實(shí)上，HTTP 請(qǐng)求可以是同步可以是異步的。一般正常的 HTTP 請(qǐng)求都是同步的，同步方式最大的一個(gè)特點(diǎn)是提交請(qǐng)求->等待服務(wù)器處理->處理完畢返回 這個(gè)期間客戶端瀏覽器不能做任何事。而異步方式最大的特點(diǎn)是 請(qǐng)求通過(guò)事件觸發(fā)->服務(wù)器處理(這時(shí)瀏覽器仍然可以做其他事情)-> 處理完畢。

那么我也可以說(shuō)同步請(qǐng)求就是順序處理的，而異步請(qǐng)求的執(zhí)行方式則不確定，因?yàn)楫惒叫枰獎(jiǎng)?chuàng)建多個(gè)執(zhí)行線程，而每個(gè)線程的執(zhí)行順序不同。

這里需要注意一點(diǎn)，我們只是使用 HTTP 請(qǐng)求來(lái)舉例子，而 Kafka 采用的是 TCP 基于 Socket 的方式進(jìn)行通訊

那么這兩種方式有什么缺點(diǎn)呢?

我相信聰明的你應(yīng)該能馬上想到，同步的方式最大的缺點(diǎn)就是吞吐量太差，資源利用率極低，由于只能順序處理請(qǐng)求，因此，每個(gè)請(qǐng)求都必須等待前一個(gè)請(qǐng)求處理完畢才能得到處理。這種方式只適用于請(qǐng)求發(fā)送非常不頻繁的系統(tǒng)。

異步的方式的缺點(diǎn)就是為每個(gè)請(qǐng)求都創(chuàng)建線程的做法開(kāi)銷(xiāo)極大，在某些場(chǎng)景下甚至?xí)嚎逭麄€(gè)服務(wù)。

響應(yīng)式模型

說(shuō)了這么半天，Kafka 采用同步還是異步的呢?都不是，Kafka 采用的是一種 響應(yīng)式(Reactor)模型，那么什么是響應(yīng)式模型呢?簡(jiǎn)單的說(shuō)，Reactor 模式是事件驅(qū)動(dòng)架構(gòu)的一種實(shí)現(xiàn)方式，特別適合應(yīng)用于處理多個(gè)客戶端并發(fā)向服務(wù)器端發(fā)送請(qǐng)求的場(chǎng)景，如下圖所示：

Kafka 的 broker 端有個(gè) SocketServer組件，類似于處理器，SocketServer 是基于 TCP 的 Socket 連接的，它用于接受客戶端請(qǐng)求，所有的請(qǐng)求消息都包含一個(gè)消息頭，消息頭中都包含如下信息：

• Request type (也就是 API Key)
• Request version(broker 可以處理不同版本的客戶端請(qǐng)求，并根據(jù)客戶版本做出不同的響應(yīng))
• Correlation ID --- 一個(gè)具有唯一性的數(shù)字，用于標(biāo)示請(qǐng)求消息，同時(shí)也會(huì)出現(xiàn)在響應(yīng)消息和錯(cuò)誤日志中(用于診斷問(wèn)題)
• Client ID --- 用于標(biāo)示發(fā)送請(qǐng)求的客戶端

broker 會(huì)在它所監(jiān)聽(tīng)的每一個(gè)端口上運(yùn)行一個(gè) Acceptor 線程，這個(gè)線程會(huì)創(chuàng)建一個(gè)連接，并把它交給 Processor(網(wǎng)絡(luò)線程池)， Processor 的數(shù)量可以使用 num.network.threads 進(jìn)行配置，其默認(rèn)值是3，表示每臺(tái) broker 啟動(dòng)時(shí)會(huì)創(chuàng)建3個(gè)線程，專門(mén)處理客戶端發(fā)送的請(qǐng)求。

Acceptor 線程會(huì)采用輪詢的方式將入棧請(qǐng)求公平的發(fā)送至網(wǎng)絡(luò)線程池中，因此，在實(shí)際使用過(guò)程中，這些線程通常具有相同的機(jī)率被分配到待處理請(qǐng)求隊(duì)列中，然后從響應(yīng)隊(duì)列獲取響應(yīng)消息，把它們發(fā)送給客戶端。Processor 網(wǎng)絡(luò)線程池中的請(qǐng)求 - 響應(yīng)的處理還是比較復(fù)雜的，下面是網(wǎng)絡(luò)線程池中的處理流程圖：

Processor 網(wǎng)絡(luò)線程池接收到客戶和其他 broker 發(fā)送來(lái)的消息后，網(wǎng)絡(luò)線程池會(huì)把消息放到請(qǐng)求隊(duì)列中，注意這個(gè)是共享請(qǐng)求隊(duì)列，因?yàn)榫W(wǎng)絡(luò)線程池是多線程機(jī)制的，所以請(qǐng)求隊(duì)列的消息是多線程共享的區(qū)域，然后由 IO 線程池進(jìn)行處理，根據(jù)消息的種類判斷做何處理，比如 PRODUCE 請(qǐng)求，就會(huì)將消息寫(xiě)入到 log 日志中，如果是FETCH請(qǐng)求，則從磁盤(pán)或者頁(yè)緩存中讀取消息。也就是說(shuō)，IO線程池是真正做判斷，處理請(qǐng)求的一個(gè)組件。在IO 線程池處理完畢后，就會(huì)判斷是放入響應(yīng)隊(duì)列中還是 Purgatory 中，Purgatory 是什么我們下面再說(shuō)，現(xiàn)在先說(shuō)一下響應(yīng)隊(duì)列，響應(yīng)隊(duì)列是每個(gè)線程所獨(dú)有的，因?yàn)轫憫?yīng)式模型中不會(huì)關(guān)心請(qǐng)求發(fā)往何處，因此把響應(yīng)回傳的事情就交給每個(gè)線程了，所以也就不必共享了。

注意：IO 線程池可以通過(guò) broker 端參數(shù) num.io.threads 來(lái)配置，默認(rèn)的線程數(shù)是8，表示每臺(tái) broker 啟動(dòng)后自動(dòng)創(chuàng)建 8 個(gè)IO 處理線程。

請(qǐng)求類型

下面是幾種常見(jiàn)的請(qǐng)求類型：

生產(chǎn)請(qǐng)求

簡(jiǎn)單來(lái)講就是不同的配置對(duì)寫(xiě)入成功的界定是不同的，如果 acks = 1，那么只要領(lǐng)導(dǎo)者收到消息就表示寫(xiě)入成功，如果acks = 0，表示只要領(lǐng)導(dǎo)者發(fā)送消息就表示寫(xiě)入成功，根本不用考慮返回值的影響。如果 acks = all，就表示領(lǐng)導(dǎo)者需要收到所有副本的消息后才表示寫(xiě)入成功。

在消息被寫(xiě)入分區(qū)的首領(lǐng)后，如果 acks 配置的值是 all，那么這些請(qǐng)求會(huì)被保存在 煉獄(Purgatory)的緩沖區(qū)中，直到領(lǐng)導(dǎo)者副本發(fā)現(xiàn)跟隨者副本都復(fù)制了消息，響應(yīng)才會(huì)發(fā)送給客戶端。

獲取請(qǐng)求

broker 獲取請(qǐng)求的方式與處理生產(chǎn)請(qǐng)求的方式類似，客戶端發(fā)送請(qǐng)求，向 broker 請(qǐng)求主題分區(qū)中特定偏移量的消息，如果偏移量存在，Kafka 會(huì)采用 零復(fù)制 技術(shù)向客戶端發(fā)送消息，Kafka 會(huì)直接把消息從文件中發(fā)送到網(wǎng)絡(luò)通道中，而不需要經(jīng)過(guò)任何的緩沖區(qū)，從而獲得更好的性能。

客戶端可以設(shè)置獲取請(qǐng)求數(shù)據(jù)的上限和下限，上限指的是客戶端為接受足夠消息分配的內(nèi)存空間，這個(gè)限制比較重要，如果上限太大的話，很有可能直接耗盡客戶端內(nèi)存。下限可以理解為攢足了數(shù)據(jù)包再發(fā)送的意思，這就相當(dāng)于項(xiàng)目經(jīng)理給程序員分配了 10 個(gè)bug，程序員每次改一個(gè) bug 就會(huì)向項(xiàng)目經(jīng)理匯報(bào)一下，有的時(shí)候改好了有的時(shí)候可能還沒(méi)改好，這樣就增加了溝通成本和時(shí)間成本，所以下限值得就是程序員你改完10個(gè) bug 再向我匯報(bào)!!!如下圖所示：

如圖你可以看到，在拉取消息 ---> 消息 之間是有一個(gè)等待消息積累這么一個(gè)過(guò)程的，這個(gè)消息積累你可以把它想象成超時(shí)時(shí)間，不過(guò)超時(shí)會(huì)跑出異常，消息積累超時(shí)后會(huì)響應(yīng)回執(zhí)。延遲時(shí)間可以通過(guò) replica.lag.time.max.ms 來(lái)配置，它指定了副本在復(fù)制消息時(shí)可被允許的最大延遲時(shí)間。

元數(shù)據(jù)請(qǐng)求

生產(chǎn)請(qǐng)求和響應(yīng)請(qǐng)求都必須發(fā)送給領(lǐng)導(dǎo)者副本，如果 broker 收到一個(gè)針對(duì)某個(gè)特定分區(qū)的請(qǐng)求，而該請(qǐng)求的首領(lǐng)在另外一個(gè) broker 中，那么發(fā)送請(qǐng)求的客戶端會(huì)收到非分區(qū)首領(lǐng)的錯(cuò)誤響應(yīng);如果針對(duì)某個(gè)分區(qū)的請(qǐng)求被發(fā)送到不含有領(lǐng)導(dǎo)者的 broker 上，也會(huì)出現(xiàn)同樣的錯(cuò)誤。Kafka 客戶端需要把請(qǐng)求和響應(yīng)發(fā)送到正確的 broker 上。這不是廢話么?我怎么知道要往哪發(fā)送?

事實(shí)上，客戶端會(huì)使用一種 元數(shù)據(jù)請(qǐng)求 ，這種請(qǐng)求會(huì)包含客戶端感興趣的主題列表，服務(wù)端的響應(yīng)消息指明了主題的分區(qū)，領(lǐng)導(dǎo)者副本和跟隨者副本。元數(shù)據(jù)請(qǐng)求可以發(fā)送給任意一個(gè) broker，因?yàn)樗械?broker 都會(huì)緩存這些信息。

一般情況下，客戶端會(huì)把這些信息緩存，并直接向目標(biāo) broker 發(fā)送生產(chǎn)請(qǐng)求和相應(yīng)請(qǐng)求，這些緩存需要隔一段時(shí)間就進(jìn)行刷新，使用metadata.max.age.ms 參數(shù)來(lái)配置，從而知道元數(shù)據(jù)是否發(fā)生了變更。比如，新的 broker 加入后，會(huì)觸發(fā)重平衡，部分副本會(huì)移動(dòng)到新的 broker 上。這時(shí)候，如果客戶端收到 不是首領(lǐng)的錯(cuò)誤，客戶端在發(fā)送請(qǐng)求之前刷新元數(shù)據(jù)緩存。

Kafka 重平衡流程

我在 真的，關(guān)于 Kafka 入門(mén)看這一篇就夠了 中關(guān)于消費(fèi)者描述的時(shí)候大致說(shuō)了一下消費(fèi)者組和重平衡之間的關(guān)系，實(shí)際上，歸納為一點(diǎn)就是讓組內(nèi)所有的消費(fèi)者實(shí)例就消費(fèi)哪些主題分區(qū)達(dá)成一致。

我們知道，一個(gè)消費(fèi)者組中是要有一個(gè)群組協(xié)調(diào)者(Coordinator)的，而重平衡的流程就是由 Coordinator 的幫助下來(lái)完成的。

這里需要先聲明一下重平衡發(fā)生的條件：

• 消費(fèi)者訂閱的任何主題發(fā)生變化
• 消費(fèi)者數(shù)量發(fā)生變化
• 分區(qū)數(shù)量發(fā)生變化
• 如果你訂閱了一個(gè)還尚未創(chuàng)建的主題，那么重平衡在該主題創(chuàng)建時(shí)發(fā)生。如果你訂閱的主題發(fā)生刪除那么也會(huì)發(fā)生重平衡
• 消費(fèi)者被群組協(xié)調(diào)器認(rèn)為是 DEAD 狀態(tài)，這可能是由于消費(fèi)者崩潰或者長(zhǎng)時(shí)間處于運(yùn)行狀態(tài)下發(fā)生的，這意味著在配置合理時(shí)間的范圍內(nèi)，消費(fèi)者沒(méi)有向群組協(xié)調(diào)器發(fā)送任何心跳，這也會(huì)導(dǎo)致重平衡的發(fā)生。

在了解重平衡之前，你需要知道這兩個(gè)角色

群組協(xié)調(diào)器(Coordinator)：群組協(xié)調(diào)器是一個(gè)能夠從消費(fèi)者群組中收到所有消費(fèi)者發(fā)送心跳消息的 broker。在最早期的版本中，元數(shù)據(jù)信息是保存在 ZooKeeper 中的，但是目前元數(shù)據(jù)信息存儲(chǔ)到了 broker 中。每個(gè)消費(fèi)者組都應(yīng)該和群組中的群組協(xié)調(diào)器同步。當(dāng)所有的決策要在應(yīng)用程序節(jié)點(diǎn)中進(jìn)行時(shí)，群組協(xié)調(diào)器可以滿足 JoinGroup 請(qǐng)求并提供有關(guān)消費(fèi)者組的元數(shù)據(jù)信息，例如分配和偏移量。群組協(xié)調(diào)器還有權(quán)知道所有消費(fèi)者的心跳，消費(fèi)者群組中還有一個(gè)角色就是領(lǐng)導(dǎo)者，注意把它和領(lǐng)導(dǎo)者副本和 kafka controller 進(jìn)行區(qū)分。領(lǐng)導(dǎo)者是群組中負(fù)責(zé)決策的角色，所以如果領(lǐng)導(dǎo)者掉線了，群組協(xié)調(diào)器有權(quán)把所有消費(fèi)者踢出組。因此，消費(fèi)者群組的一個(gè)很重要的行為是選舉領(lǐng)導(dǎo)者，并與協(xié)調(diào)器讀取和寫(xiě)入有關(guān)分配和分區(qū)的元數(shù)據(jù)信息。

消費(fèi)者領(lǐng)導(dǎo)者：每個(gè)消費(fèi)者群組中都有一個(gè)領(lǐng)導(dǎo)者。如果消費(fèi)者停止發(fā)送心跳了，協(xié)調(diào)者會(huì)觸發(fā)重平衡。

在了解重平衡之前，你需要知道狀態(tài)機(jī)是什么

Kafka 設(shè)計(jì)了一套消費(fèi)者組狀態(tài)機(jī)(State Machine) ，來(lái)幫助協(xié)調(diào)者完成整個(gè)重平衡流程。消費(fèi)者狀態(tài)機(jī)主要有五種狀態(tài)它們分別是 Empty、Dead、PreparingRebalance、CompletingRebalance 和 Stable。

了解了這些狀態(tài)的含義之后，下面我們用幾條路徑來(lái)表示一下消費(fèi)者狀態(tài)的輪轉(zhuǎn)

1.消費(fèi)者組一開(kāi)始處于 Empty 狀態(tài)，當(dāng)重平衡開(kāi)啟后，它會(huì)被置于 PreparingRebalance狀態(tài)等待新消費(fèi)者的加入，一旦有新的消費(fèi)者加入后，消費(fèi)者群組就會(huì)處于 CompletingRebalance 狀態(tài)等待分配，只要有新的消費(fèi)者加入群組或者離開(kāi)，就會(huì)觸發(fā)重平衡，消費(fèi)者的狀態(tài)處于 PreparingRebalance 狀態(tài)。等待分配機(jī)制指定好后完成分配，那么它的流程圖是這樣的。

2.在上圖的基礎(chǔ)上，當(dāng)消費(fèi)者群組都到達(dá) Stable 狀態(tài)后，一旦有新的消費(fèi)者加入/離開(kāi)/心跳過(guò)期，那么觸發(fā)重平衡，消費(fèi)者群組的狀態(tài)重新處于 PreparingRebalance 狀態(tài)。那么它的流程圖是這樣的。

3.在上圖的基礎(chǔ)上，消費(fèi)者群組處于 PreparingRebalance 狀態(tài)后，很不幸，沒(méi)人玩兒了，所有消費(fèi)者都離開(kāi)了，這時(shí)候還可能會(huì)保留有消費(fèi)者消費(fèi)的位移數(shù)據(jù)，一旦位移數(shù)據(jù)過(guò)期或者被刷新，那么消費(fèi)者群組就處于 Dead 狀態(tài)了。它的流程圖是這樣的：

4.在上圖的基礎(chǔ)上，我們分析了消費(fèi)者的重平衡，在 PreparingRebalance或者 CompletingRebalance 或者 Stable 任意一種狀態(tài)下發(fā)生位移主題分區(qū) Leader 發(fā)生變更，群組會(huì)直接處于 Dead 狀態(tài)，它的所有路徑如下：

這里面需要注意兩點(diǎn)：

一般出現(xiàn) Required xx expired offsets in xxx milliseconds 就表明Kafka 很可能就把該組的位移數(shù)據(jù)刪除了

只有 Empty 狀態(tài)下的組，才會(huì)執(zhí)行過(guò)期位移刪除的操作。

重平衡流程

上面我們了解到了消費(fèi)者群組狀態(tài)的轉(zhuǎn)化過(guò)程，下面我們真正開(kāi)始介紹 Rebalance 的過(guò)程。重平衡過(guò)程可以從兩個(gè)方面去看：消費(fèi)者端和協(xié)調(diào)者端，首先我們先看一下消費(fèi)者端

從消費(fèi)者看重平衡

從消費(fèi)者看重平衡有兩個(gè)步驟：分別是 消費(fèi)者加入組 和 等待領(lǐng)導(dǎo)者分配方案。這兩個(gè)步驟后分別對(duì)應(yīng)的請(qǐng)求是 JoinGroup 和 SyncGroup。

新的消費(fèi)者加入群組時(shí)，這個(gè)消費(fèi)者會(huì)向協(xié)調(diào)器發(fā)送 JoinGroup 請(qǐng)求。在該請(qǐng)求中，每個(gè)消費(fèi)者成員都需要將自己消費(fèi)的 topic 進(jìn)行提交，我們上面描述群組協(xié)調(diào)器中說(shuō)過(guò)，這么做的目的就是為了讓協(xié)調(diào)器收集足夠的元數(shù)據(jù)信息，來(lái)選取消費(fèi)者組的領(lǐng)導(dǎo)者。通常情況下，第一個(gè)發(fā)送 JoinGroup 請(qǐng)求的消費(fèi)者會(huì)自動(dòng)稱為領(lǐng)導(dǎo)者。領(lǐng)導(dǎo)者的任務(wù)是收集所有成員的訂閱信息，然后根據(jù)這些信息，制定具體的分區(qū)消費(fèi)分配方案。如圖：

在所有的消費(fèi)者都加入進(jìn)來(lái)并把元數(shù)據(jù)信息提交給領(lǐng)導(dǎo)者后，領(lǐng)導(dǎo)者做出分配方案并發(fā)送 SyncGroup請(qǐng)求給協(xié)調(diào)者，協(xié)調(diào)者負(fù)責(zé)下發(fā)群組中的消費(fèi)策略。下圖描述了 SyncGroup 請(qǐng)求的過(guò)程。

當(dāng)所有成員都成功接收到分配方案后，消費(fèi)者組進(jìn)入到 Stable 狀態(tài)，即開(kāi)始正常的消費(fèi)工作。

從協(xié)調(diào)者來(lái)看重平衡

從協(xié)調(diào)者角度來(lái)看重平衡主要有下面這幾種觸發(fā)條件，

新成員加入組

組成員主動(dòng)離開(kāi)

組成員崩潰離開(kāi)

組成員提交位移

我們分別來(lái)描述一下，先從新成員加入組開(kāi)始

新成員入組

我們討論的場(chǎng)景消費(fèi)者集群狀態(tài)處于Stable 等待分配的過(guò)程，這時(shí)候如果有新的成員加入組的話，重平衡的過(guò)程。

從這個(gè)角度來(lái)看，協(xié)調(diào)者的過(guò)程和消費(fèi)者類似，只是剛剛從消費(fèi)者的角度去看，現(xiàn)在從領(lǐng)導(dǎo)者的角度去看

組成員離開(kāi)

組成員離開(kāi)消費(fèi)者群組指的是消費(fèi)者實(shí)例調(diào)用 close() 方法主動(dòng)通知協(xié)調(diào)者它要退出。這里又會(huì)有一個(gè)新的請(qǐng)求出現(xiàn) LeaveGroup()請(qǐng)求 。如下圖所示：

組成員崩潰

組成員崩潰是指消費(fèi)者實(shí)例出現(xiàn)嚴(yán)重故障，宕機(jī)或者一段時(shí)間未響應(yīng)，協(xié)調(diào)者接收不到消費(fèi)者的心跳，就會(huì)被認(rèn)為是組成員崩潰，崩潰離組是被動(dòng)的，協(xié)調(diào)者通常需要等待一段時(shí)間才能感知到，這段時(shí)間一般是由消費(fèi)者端參數(shù) session.timeout.ms 控制的。如下圖所示：

重平衡時(shí)提交位移

這個(gè)過(guò)程我們就不再用圖形來(lái)表示了，大致描述一下就是 消費(fèi)者發(fā)送 JoinGroup 請(qǐng)求后，群組中的消費(fèi)者必須在指定的時(shí)間范圍內(nèi)提交各自的位移，然后再開(kāi)啟正常的 JoinGroup/SyncGroup 請(qǐng)求發(fā)送。

文章參考：

• 《Kafka 權(quán)威指南》
• https://blog.csdn.net/u013256816/article/details/80546337
• https://learning.oreilly.com/library/view/kafka-the-definitive/9781491936153/ch05.html#kafka_internals
• https://www.cnblogs.com/kevingrace/p/9021508.html
• https://www.cnblogs.com/huxi2b/p/6980045.html
• 《極客時(shí)間-Kafka核心技術(shù)與實(shí)戰(zhàn)》
• https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Controller+Redesign
• https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Controller+Internals
• kafka 分區(qū)和副本以及kafaka 執(zhí)行流程，以及消息的高可用
• Http中的同步請(qǐng)求和異步請(qǐng)求
• Reactor模式詳解
• https://kafka.apache.org/documentation/
• https://www.linkedin.com/pulse/partitions-rebalance-kafka-raghunandan-gupta
• https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Detailed+Consumer+Coordinator+Design

 【編輯推薦】

1. 如何驗(yàn)證 Kafka 系統(tǒng)的可靠性？這下我終于懂了
2. Zookeeper的選舉算法和腦裂問(wèn)題深度講解
3. 開(kāi)源Apache Cassandra、Kafka、Spark和ES何時(shí)該用，何時(shí)不該用？
4. 每秒幾十億實(shí)時(shí)處理，阿里巴巴超大規(guī)模 Flink 集群運(yùn)維揭秘
5. 每天處理千億級(jí)日志量，Kafka是如何做到的?

分享標(biāo)題：你能說(shuō)出Kafka這些原理嗎
網(wǎng)頁(yè)地址：http://m.fisionsoft.com.cn/article/cddgigd.html