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

[[287183]]

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

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

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

下面回到技術(shù)。

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

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

如果感興趣的話，就請花費你一些時間，耐心看完這篇文章。

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

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

Kafka 可以通過 broker.id.generation.enable 和 reserved.broker.max.id 來配合生成新的 broker.id。

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

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

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

Broker Controller 的作用

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

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

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

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

控制器的選舉

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

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

控制器的作用

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

broker controller 故障轉(zhuǎn)移

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

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

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

broker controller 存在的問題

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

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

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

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

主要所做的改變有下面這幾點：

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

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

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

副本機制

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

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

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

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

Leader 副本

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

Follower 副本

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

這幅圖需要注意以下幾點：

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

首領(lǐng)的另一個任務(wù)是搞清楚哪個跟隨者的狀態(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ù)的請求，這種請求與消費者為了讀取消息而發(fā)送的信息是一樣的。

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

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

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

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

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

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

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

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

同步復(fù)制

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

異步復(fù)制

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

ISR

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

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

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

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

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

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

Kafka 請求處理流程

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

那么我也可以說同步請求就是順序處理的，而異步請求的執(zhí)行方式則不確定，因為異步需要創(chuàng)建多個執(zhí)行線程，而每個線程的執(zhí)行順序不同。

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

那么這兩種方式有什么缺點呢?

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

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

響應(yīng)式模型

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

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

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

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

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

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

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

請求類型

下面是幾種常見的請求類型：

生產(chǎn)請求

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

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

獲取請求

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

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

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

元數(shù)據(jù)請求

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

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

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

Kafka 重平衡流程

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

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

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

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

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

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

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

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

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

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

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

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

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

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

這里面需要注意兩點：

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

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

重平衡流程

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

從消費者看重平衡

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

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

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

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

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

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

新成員加入組

組成員主動離開

組成員崩潰離開

組成員提交位移

我們分別來描述一下，先從新成員加入組開始

新成員入組

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

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

組成員離開

組成員離開消費者群組指的是消費者實例調(diào)用 close() 方法主動通知協(xié)調(diào)者它要退出。這里又會有一個新的請求出現(xiàn) LeaveGroup()請求 。如下圖所示：

組成員崩潰

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

重平衡時提交位移

這個過程我們就不再用圖形來表示了，大致描述一下就是 消費者發(fā)送 JoinGroup 請求后，群組中的消費者必須在指定的時間范圍內(nèi)提交各自的位移，然后再開啟正常的 JoinGroup/SyncGroup 請求發(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
• 《極客時間-Kafka核心技術(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中的同步請求和異步請求
• 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

 【編輯推薦】