[TOC]
6.1 協(xié)議設(shè)計(jì)
在實(shí)際應(yīng)用中步咪， Kafka 經(jīng)常被用作高性能妇菱、可擴(kuò)展的消息中間件 务荆。 Kafka 自定義了 一組基于 TCP 的二進(jìn)制協(xié)議，只要遵守這組協(xié)議的格式校坑，就可以向 Kafka 發(fā)送消息拣技，也可以從 Kafka 中 拉取消息，或者做一些其他的事情耍目，比如提交消費(fèi)位移等膏斤。
在目前的 Kafka 2.0.0 中， 一共包含了 43 種協(xié)議類型邪驮，每種協(xié)議類型都有對(duì)應(yīng)的請(qǐng)求 (Request)和響應(yīng) Response)莫辨，它們都遵守特定的協(xié)議模式。每種類型的 Request 都包含相同 結(jié)構(gòu)的協(xié)議請(qǐng)求頭( RequestHeader)和不同結(jié)構(gòu)的協(xié)議請(qǐng)求體 CRequestBody)毅访，如圖 6-1 所示沮榜。
協(xié)議請(qǐng)求頭中包含 4 個(gè)域( Field) : api key、 api_version喻粹、 correlation id 和client_id
每種類型的 Response 也包含相同結(jié)構(gòu)的協(xié)議響應(yīng)頭( ResponseHeader)和不同結(jié)構(gòu)的響應(yīng) 體(ResponseBody) 蟆融，如圖 6-2所示。
協(xié)議響應(yīng)頭中只有 一個(gè) correlation id守呜，對(duì)應(yīng)的釋義可以參考表 6-1 中 的相關(guān)描述 型酥。
細(xì)心的讀者會(huì)發(fā)現(xiàn)不管是在圖 6-1 中還是在圖 6-2 中都有類似 int32、 int16查乒、 string 的字樣弥喉， 它們用 來(lái)表示當(dāng)前域的數(shù)據(jù)類型 。 Kafka 中所有協(xié)議類型的 Request 和 Response 的結(jié)構(gòu)都是具 備固定格式的玛迄，并且它 們 都構(gòu)建于多種基本數(shù)據(jù)類型之上 由境。 這些基本數(shù)據(jù)類型如圖 6-2 所示。
下面就 以最常見的消息發(fā)送和消息拉取的兩種協(xié)議類型做細(xì)致的講解蓖议。首先要講述的是消 息發(fā)送的協(xié)議類型虏杰，即 ProduceRequest/ProduceResponse讥蟆，對(duì)應(yīng)的 api_key= 0，表示 PRODUCE嘹屯。 從Kafka建立之初， 其所支持的協(xié)議類型就一直在增加从撼， 并且對(duì)特定的協(xié)議類型而言州弟，內(nèi)部的 組織結(jié)構(gòu)也并非一成不變。 以 ProduceRequest/ ProduceResponse 為例低零， 截至 目前就經(jīng)歷了 7 個(gè) 版本(VO~V6) 的變遷婆翔。 下面就以最新版本 CV6， 即api_version=6) 的結(jié)構(gòu)為例來(lái)做細(xì)致的 講解掏婶。 ProduceRequest 的組織結(jié)構(gòu)如圖 6-3 所示啃奴。
除了請(qǐng)求頭中的 4個(gè)域， 其余 ProduceRequest請(qǐng)求體中各個(gè)域的含義如表 6-3 所示雄妥。
在 2.2.l 節(jié)中我們了解到:消息累加器 RecordAccumulator 中的消息是以<分區(qū)最蕾， Deque< ProducerBatch>>的形式進(jìn)行緩存的，之后由 Sender線程轉(zhuǎn)變成<Node, List<ProducerBatch>>的 形式老厌，針對(duì)每個(gè) Node, Sender線程在發(fā)送消息前會(huì)將對(duì)應(yīng)的 List<ProducerBatch>形式的內(nèi)容轉(zhuǎn) 變成 ProduceRequest 的具體結(jié)構(gòu) 瘟则。 List<ProducerBatch>中 的內(nèi)容首先會(huì)按照主題名稱進(jìn)行分類(對(duì)應(yīng) ProduceRequest 中的域 topic)，然后按照分區(qū)編號(hào)進(jìn)行分類(對(duì)應(yīng) ProduceRequest 中 的域 partition)枝秤，分類之后的 ProducerBatch集合就對(duì)應(yīng) ProduceRequest中的域 record set醋拧。 從另 一個(gè)角度來(lái)講 ， 每個(gè)分區(qū)中的消息是順序追加的 淀弹， 那么在客戶端中按照分區(qū)歸納好之后就 可以省去在服務(wù)端 中轉(zhuǎn)換的操作了 丹壕， 這樣將負(fù)載的壓力分?jǐn)偨o了客戶端，從而使服務(wù)端可以專 注于它的分內(nèi)之事薇溃，如此也可以提升 整體 的性能 菌赖。
除了響應(yīng)頭中的 correlation_id，其余 ProduceResponse各個(gè)域的含義如表 6-4所示沐序。
我們?cè)賮?lái)了解一下拉取消息的協(xié)議類型盏袄，即 FetchRequest/FetchResponse，對(duì)應(yīng)的 api_key= 1, 表示 FETCH薄啥。 截至目前辕羽， FetchRequest/FetchResponse 一共歷經(jīng)了 9 個(gè)版本 (VO~V8)的變遷， 下面就以最新版本 (V8)的結(jié)構(gòu)為例來(lái)做細(xì)致的講解垄惧。 FetchRequest的組織結(jié)構(gòu)如圖 6-5所示刁愿。
除了請(qǐng)求頭中的 4個(gè)域，其余 FetchRequest中各個(gè)域的含義如表 6-5所示到逊。
不管是 follower 副本還是普通的消費(fèi)者客戶端铣口，如果要拉取某個(gè)分區(qū)中的消息滤钱，就需要指 定詳細(xì)的拉取信息， 也就是需要設(shè)定 partit工on脑题、 fetch offset件缸、 log start offset 和max bytes這4個(gè)域的具體值， 那么對(duì)每個(gè)分區(qū)而言叔遂，就需要占用4B+8B+8B+4B=24B的 空間 他炊。 一般情況下，不管是 follower 副本還是普通的消費(fèi)者已艰，它們的訂閱信息是長(zhǎng)期固定的痊末。 也就是說(shuō)， FetchRequest 中的 topics 域的內(nèi)容是長(zhǎng)期固定的哩掺，只有在拉取開始時(shí)或發(fā)生某些 異常時(shí)會(huì)有所變動(dòng) 凿叠。 FetchRequest 請(qǐng)求是一個(gè)非常頻繁的請(qǐng)求，如果要拉取的分區(qū)數(shù)有很多嚼吞，比如有 1000個(gè)分區(qū)盒件，那么在網(wǎng)絡(luò)上頻繁交互 FetchRequest時(shí)就會(huì)有固定的 1000×24B ~ 24KB 的字節(jié)的內(nèi)容在傳動(dòng)，如果可以將這 24陽(yáng)的狀態(tài)保存起來(lái)舱禽，那么就可以節(jié)省這部分所占用的 帶寬履恩。
Kafka 從 1.1.0 版本開始針對(duì) FetchRequest 引入了 session_id、 epoch 和 forgotten topics_data等域呢蔫， session_id和epoch確定一條拉取鏈路的fetchsession切心，當(dāng)session建 立或變更時(shí)會(huì)發(fā)送全量式的 FetchRequest，所謂的全量式就是指請(qǐng)求體中包含所有需要拉取 的 分區(qū)信息 : 當(dāng) session 穩(wěn)定時(shí)則會(huì)發(fā)送增量式的 FetchRequest 請(qǐng)求片吊，里面的 topics 域?yàn)榭?绽昏，因 為 topics 域的內(nèi)容己經(jīng)被緩存在了 session 鏈路的兩側(cè)。如果需要從當(dāng)前 fetch session 中取消 對(duì)某些分區(qū)的拉取訂閱俏脊，則可以使用 forgotten topics data 字段來(lái)實(shí)現(xiàn)全谤。
這個(gè)改進(jìn)在大規(guī)模(有大量的分區(qū)副本需要及時(shí)同步)的 Kafka集群中非常有用，它可以 提升集群間的網(wǎng)絡(luò)帶寬的有效使用率爷贫。不過(guò)對(duì)客戶端而言效果不是那么明顯认然，一般情況下單個(gè) 客戶端不會(huì)訂閱太多的分區(qū)，不過(guò)總體上這也是一個(gè)很好的優(yōu)化改進(jìn)漫萄。
與 FetchRequest對(duì)應(yīng)的 FetchResponse 的組織結(jié)構(gòu) CV8 版本)可以參考圖 6-6卷员。
FetchResponse結(jié)構(gòu)中的域也很多，它主要分為 4層腾务，第 l 層包含 throttle time ms毕骡、 error_code、 session_id 和 responses，前面 3 個(gè)域都見過(guò)未巫，其中 session_id 和 FetchRequest 中的 session id 對(duì)應(yīng)窿撬。 responses 是一個(gè)數(shù)組類型，表示響應(yīng)的具體內(nèi)容叙凡， 也就是 FetchResponse 結(jié)構(gòu)中的第 2 層劈伴，具體地細(xì)化到每個(gè)分區(qū)的響應(yīng)。第 3 層中包含分區(qū)的元 數(shù)據(jù)信息( partition 握爷、 error code 等)及具體的消息 內(nèi) 容( record set ) aborted_transactions 和事務(wù)相關(guān)跛璧。
除了 Kafka 客戶端開發(fā)人員，絕大多數(shù)的其他開發(fā)人員基本接觸不到或不需要接觸具體的 協(xié)議饼拍，那么我們?yōu)槭裁催€要了解它們呢?其實(shí)赡模，協(xié)議的具體定義可以讓我們從另一個(gè)角度來(lái)了 解 Kafka 的本質(zhì) 田炭。以 PRODUCE 和 FETCH 為例师抄，從協(xié)議 結(jié)構(gòu)中就可 以看出消息 的 寫入和拉取 消費(fèi)都是細(xì)化到每 一個(gè)分區(qū)層級(jí)的。并且教硫，通過(guò)了解各個(gè)協(xié)議版本變遷的細(xì)節(jié)也能夠從側(cè)面了 解 Kafka 變遷的歷史叨吮，在變遷的過(guò)程中遇到 過(guò)哪方面的瓶頸， 又采取哪種優(yōu) 化手段瞬矩，比如 FetchRequest 中的 session_id 的引 入 茶鉴。
6.2 時(shí)間輪
Kafka中存在大量的延時(shí)操作，比如延時(shí)生產(chǎn)景用、延時(shí)拉取和延時(shí)刪除等涵叮。 Kafka并沒有使用 JDK 自帶的 Timer 或 DelayQueue 來(lái)實(shí)現(xiàn)延時(shí)的功能，而是基于時(shí)間輪的概念自定義實(shí)現(xiàn)了一個(gè) 用于延時(shí)功能的定時(shí)器( SystemTimer)伞插。 JDK 中 Timer 和 DelayQueue 的插入和刪除操作的平 均時(shí)間復(fù)雜度為 O(nlogn)并不能滿足 Kafka 的高性能要求割粮，而基于時(shí)間輪可以將插入和刪除操 作的時(shí)間復(fù)雜度都降為 0(1)。 時(shí)間輪的應(yīng)用并非 Kafka獨(dú)有媚污，其應(yīng)用場(chǎng)景還有很多舀瓢，在 Netty、 Akka, Quartz耗美、 ZooKeeper 等組件中都存在時(shí)間輪的蹤影 京髓。
如圖 6-7 所示， Kafka 中的時(shí)間輪( TimingWheel)是一個(gè)存儲(chǔ)定時(shí)任務(wù)的環(huán)形隊(duì)列 商架， 底層 采用數(shù)組實(shí)現(xiàn)堰怨，數(shù)組中的每個(gè)元素可以存放一個(gè)定時(shí)任務(wù)列表( TimerTaskList)。 TimerTaskList 是一個(gè)環(huán)形的雙向鏈表蛇摸，鏈表中的每一項(xiàng)表示的都是定時(shí)任務(wù)項(xiàng)( TimerTaskEntry)诚些，其中封裝了真正的定時(shí)任務(wù) (TimerTask) 。
時(shí)間輪由多個(gè)時(shí)間格組成， 每個(gè)時(shí) 間格代表 當(dāng)前時(shí)間輪的基本時(shí)間跨度( tic燦ifs) 诬烹。時(shí) 間 輪的時(shí)間格個(gè)數(shù)是固定的砸烦，可用 wheelSize 來(lái)表示，那么整個(gè)時(shí)間輪的總體時(shí)間跨度( interval) 可以通過(guò)公式 tic燦ifs×wheelSize計(jì)算得出绞吁。 時(shí)間輪還有一個(gè)表盤指針(currentTime)幢痘，用來(lái)表 示時(shí)間輪當(dāng)前所處的時(shí)間， currentTime 是 tic燦ifs 的整數(shù)倍 家破。 currentTime 可以將整個(gè)時(shí)間輪劃分 為到期部分和未到期部分颜说， currentTime 當(dāng)前指向的時(shí)間格也屬于到期部分，表示剛好到期汰聋，需 要處理此時(shí)間格所對(duì)應(yīng)的 TimerTaskList 中的所有任務(wù)门粪。
若時(shí)間輪的 tic燦也為 lms 且 wheelSize 等于 20，那么可以計(jì)算得出總體時(shí)間跨度 interval 為 20msa 初始情況下表盤指針 currentTime 指向時(shí)間格 0烹困，此時(shí)有一個(gè)定時(shí)為 2ms 的任務(wù)插進(jìn) 來(lái)會(huì)存放到時(shí)間格為 2 的 TimerTaskList 中 玄妈。 隨著時(shí)間的不斷推移 ， 指針 currentTime 不斷向 前 推進(jìn)髓梅，過(guò)了 2ms 之后拟蜻，當(dāng)?shù)竭_(dá)時(shí)間格 2 時(shí)昆禽，就需要將時(shí)間格 2 對(duì)應(yīng)的 TimeTaskList 中的任務(wù)進(jìn) 行相應(yīng)的到期操作采桃。此時(shí)若又有一個(gè)定時(shí)為 8ms 的任務(wù)插進(jìn)來(lái)算途，則會(huì)存放到時(shí)間格 10 中色罚， currentTime再過(guò) 8ms后會(huì)指向時(shí)間格 10喷屋。 如果同時(shí)有一個(gè)定時(shí)為 19ms 的任務(wù)插進(jìn)來(lái)怎么辦? 新來(lái)的 TimerTaskEntry 會(huì)復(fù)用原來(lái)的 TimerTaskList格二，所以它會(huì)插入原本己經(jīng)到期的時(shí)間格 l践啄。 總之羡棵，整個(gè)時(shí)間輪的總體跨度是不變的蟋字，隨著指針 currentTim巳的不斷推進(jìn)稿蹲，當(dāng)前時(shí)間輪所能處 理的時(shí)間段也在不斷后移，總體時(shí)間范圍在 currentTime 和 currentTime+interval 之間 愉老。
如果此時(shí)有一個(gè)定時(shí)為 350ms 的任務(wù)該如何處理?直接擴(kuò)充 wheelSize 的大小? Kafka 中不 乏幾萬(wàn)甚至幾十萬(wàn)毫秒的定時(shí)任務(wù)场绿，這個(gè) wheelSize 的擴(kuò)充沒有底線，就算將所有的定時(shí)任務(wù)的 到期時(shí)間都設(shè)定一個(gè)上限嫉入，比如 100 萬(wàn)毫秒焰盗，那么這個(gè) wheelSize為 100 萬(wàn)毫秒的時(shí)間輪不僅占 用很大的內(nèi)存空間，而且也會(huì)拉低效率 咒林。 Kafka 為此引入了層級(jí)時(shí)間輪的概念熬拒，當(dāng)任務(wù)的到期 時(shí)間超過(guò)了當(dāng)前時(shí)間輪所表示的時(shí)間范圍時(shí)，就會(huì)嘗試添加到上層時(shí)間輪中 垫竞。
如圖 6-8 所示澎粟，復(fù)用之前的案例蛀序，第一層的時(shí)間輪 tic燦也=lms、whee!Size=20活烙、inte凹al=20ms徐裸。 第二層的時(shí)間輪的 tic刷s為第一層時(shí)間輪的 interval，即 20ms啸盏。 每一層時(shí)間輪的 whee!Size是固 定的重贺，都是 20， 那么第二層的時(shí)間輪的總體時(shí)間跨度 interval 為 400ms回懦。 以此類推气笙，這個(gè) 400ms 也是第三層的 tickMs 的大小， 第三層的時(shí)間輪的總體時(shí) 間跨度為 8000ms怯晕。
對(duì)于之前所說(shuō)的 350ms 的定時(shí)任務(wù)潜圃，顯然第一層時(shí)間輪不能滿足條件，所以就升級(jí)到第二 層時(shí) 間輪中舟茶， 最終被插入第二層時(shí)間輪中時(shí)間格 17 所對(duì)應(yīng)的 TimerTaskList谭期。如果此時(shí)又有一個(gè) 定時(shí)為 450ms 的任務(wù)，那么顯然第二層時(shí)間輪也無(wú)法滿足條件稚晚，所以又升級(jí)到第三層時(shí)間輪中崇堵， 最終被插入第三層時(shí)間輪中時(shí)間格 l 的 TimerTaskList型诚。 注意到在到期時(shí)間為[400ms,800ms)區(qū)間
內(nèi)的多個(gè)任務(wù)(比如 446ms客燕、 455ms 和 473ms 的定時(shí)任務(wù))都會(huì)被放入第 三層 時(shí)間輪的時(shí)間格1，時(shí)間格 I 對(duì)應(yīng)的 TimerTaskList 的超時(shí)時(shí)間為 400ms狰贯。 隨著時(shí)間的流逝也搓，當(dāng)此 TimerTaskList 到期之時(shí)，原本定時(shí)為 450ms 的任務(wù)還剩下 50ms 的時(shí)間涵紊，還不能執(zhí)行這個(gè)任務(wù)的到期操作 傍妒。 這里就有一個(gè)時(shí)間輪 降級(jí)的操作 ， 會(huì)將這個(gè)剩余時(shí)間為 50ms 的定時(shí)任務(wù)重新提交到層級(jí)時(shí)間 輪中摸柄，此時(shí)第一層時(shí)間輪的總體時(shí)間跨度不夠 颤练，而第二層足夠，所以該任務(wù)被放到第二層時(shí) 間 輪到期時(shí)間為[40ms,60ms)的時(shí)間格中驱负。 再經(jīng)歷40ms之后嗦玖，此時(shí)這個(gè)任務(wù)又被“察覺”，不過(guò) 還剩余 lOms跃脊，還是不能立即執(zhí)行到期操作 宇挫。 所以還要再有一次時(shí)間輪的降級(jí)，此任務(wù)被添加到 第一層時(shí)間輪到期時(shí)間為[1Oms,11ms)的時(shí)間格中酪术，之后再經(jīng)歷 lOms后器瘪，此任務(wù)真正到期，最 終執(zhí)行相應(yīng)的到期操作 。
設(shè) 計(jì) 源于生活橡疼。我 們 常見的鐘表就是一種具有 三 層結(jié)構(gòu)的時(shí)間輪援所，第一層時(shí)間輪 tic陸也=lms、 whee1Size=60欣除、 interval=1min任斋，此為秒鐘 : 第二層 tic燦1s=lmin、 wh巳e1Size=60耻涛、 interval=1hour，此為分鐘; 第三層 tickMs=1hour抹缕、 wheelSize=12澈蟆、 interval=12hours，此為時(shí)鐘卓研。
6.3 延時(shí)操作
如果在使用生產(chǎn)者客戶端發(fā)送消息 的時(shí)候?qū)?acks 參數(shù)設(shè)置為一1趴俘，那么就意味著需要等待ISR 集合 中的所有副 本都確認(rèn)收到消息之后才能 正確地收到響 應(yīng) 的結(jié) 果，或者捕 獲超時(shí)異常 奏赘。
如圖 6-9寥闪、圖 6-10 和 圖 6-1l 所示，假設(shè)某個(gè)分區(qū)有 3 個(gè)副本: leader磨淌、 follower! 和 follower2, 它們都在分區(qū)的 ISR集合中疲憋。 為了簡(jiǎn)化說(shuō)明，這里我們不考慮 ISR集合伸縮的情況梁只。 Kafka在 收到客戶端的生產(chǎn)請(qǐng)求(ProduceRequest)后缚柳，將消息 3和消息 4寫入 leader副本的本地日志文 件 。 由于客戶端設(shè)置 了 acks 為一1搪锣， 那么需要等 到 follower! 和 follower2 兩個(gè)副本都收到消息 3 和消 息 4 后才能告知客戶端正確地接收了所發(fā)送的消息 秋忙。 如果在 一 定 的時(shí)間內(nèi)， follower! 副本 或 follower2 副本沒能 夠完全拉取 到消 息 3 和消息 4构舟，那么就需要返 回超時(shí)異常給客戶端 灰追。生產(chǎn) 請(qǐng)求的超時(shí)時(shí)間由 參數(shù) request . timeout .ms 配置，默認(rèn)值為 30000狗超，即 30s弹澎。
那么這里 等待消息 3 和消息 4 寫入 followerl 副本和 follower2 副本，井返回相應(yīng)的響應(yīng)結(jié) 果給 客戶端 的動(dòng)作是由誰(shuí) 來(lái)執(zhí)行的呢?在將消息寫入 leader 副本的本地日志文件之后抡谐， Kafka 會(huì)創(chuàng)建一個(gè)延時(shí)的生產(chǎn)操作( DelayedProduce)裁奇，用來(lái)處理消息正常寫入所有副本或超時(shí)的情況， 以返回相應(yīng)的響應(yīng)結(jié)果給客戶端麦撵。
在 Kafka 中有多種延時(shí)操作刽肠，比如前面提及的延時(shí)生產(chǎn)溃肪，還有延時(shí)拉取( DelayedFetch)、 延時(shí)數(shù)據(jù)刪除( DelayedD巳leteRecords)等 音五。 延時(shí)操作需要延時(shí)返回響應(yīng)的結(jié)果惫撰，首先它必須有 一個(gè)超時(shí)時(shí)間( delayMs)，如果在這個(gè)超時(shí)時(shí)間內(nèi) 沒有完成既定的任務(wù)躺涝，那么就需要強(qiáng)制完成 以返回響應(yīng)結(jié)果給客戶端 厨钻。其次 ，延時(shí)操作不同于定時(shí)操作坚嗜，定時(shí)操作是指在特定時(shí)間之后執(zhí) 行的操作夯膀，而延時(shí)操作可以在所設(shè)定的超時(shí)時(shí)間之前完成，所以延時(shí)操作能夠支持外部事件的 觸發(fā)苍蔬。就延時(shí)生產(chǎn)操作而言诱建，它的外部事件是所要寫入消息的某個(gè)分區(qū)的 HW (高水位)發(fā)生 增長(zhǎng)。也就是說(shuō)碟绑，隨著 follower副本不斷地與 leader副本進(jìn)行消息同步俺猿，進(jìn)而促使 HW進(jìn)一步 增長(zhǎng)， HW 每增長(zhǎng)-次都會(huì)檢測(cè)是否能夠完成此次延時(shí)生產(chǎn)操作格仲，如果可以就執(zhí)行以此返回響 應(yīng)結(jié)果給客戶端;如果在超時(shí)時(shí)間內(nèi)始終無(wú)法完成押袍，則強(qiáng)制執(zhí)行 。
延時(shí)操作創(chuàng)建之后會(huì)被加入延時(shí)操作管理器( DelayedOperationPurgatory)來(lái)做專 門 的處理凯肋。 延時(shí)操作有可能會(huì)超時(shí)谊惭，每個(gè)延時(shí)操作管理器都會(huì)配備一個(gè)定時(shí)器( SystemTimer)來(lái)做超時(shí)管 理 ， 定時(shí)器的底層就是采用時(shí)間輪( TimingWheel)實(shí)現(xiàn)的 否过。 在 6.2 節(jié)中提及時(shí)間輪的輪轉(zhuǎn)是靠“收割機(jī)”線程 ExpiredOperationReap巳r來(lái)驅(qū)動(dòng)的午笛，這里的“收割機(jī)”線程就是由延時(shí)操作管理 器啟動(dòng)的惭蟋。 也就是說(shuō)苗桂，定時(shí)器、 “收割機(jī)”線程和延時(shí)操作管理器都是一一對(duì)應(yīng)的告组。 延時(shí)操作 需要支持外部事件的觸發(fā)煤伟，所以還要配備 一個(gè)監(jiān)聽池來(lái)負(fù)責(zé)監(jiān)聽每個(gè)分區(qū)的外部事件一一查看 是否有分區(qū)的 HW 發(fā)生了增長(zhǎng) 。 另外需要補(bǔ)充的是木缝，ExpiredOperationReaper 不僅可以推進(jìn)時(shí)間 輪便锨，還會(huì)定期清理監(jiān)昕池中己 完成的延時(shí)操作。
圖 6-12 描繪了客戶端在請(qǐng)求寫入消息到收到響應(yīng)結(jié)果的過(guò)程中與延時(shí)生產(chǎn)操作相關(guān)的細(xì) 節(jié)我碟， 在了解相關(guān)的概念之后應(yīng)該比較容易理解: 如果客戶端設(shè)置的 acks 參數(shù)不為一1放案，或者沒 有成功的消息寫入，那么就直接返回結(jié)果給客戶端矫俺，否 則 就需要?jiǎng)?chuàng)建延時(shí)生產(chǎn)操作并存入延時(shí) 操作管理器吱殉，最終要么由外部事件觸發(fā)掸冤，要么由超 時(shí)觸發(fā)而執(zhí)行 。
有延時(shí)生產(chǎn)就有延時(shí)拉取友雳。 以圖6-13為例稿湿，兩個(gè)folower副本都己經(jīng)拉取到了leader副本的最新位置，此時(shí)又向 leader副本發(fā)送拉取請(qǐng)求押赊，而 leader副本并沒有新的消息寫入饺藤，那么此 時(shí) leader 副本該如何處理呢?可以 直接返回空的拉取結(jié)果給 follower 副本，不過(guò)在 lead巳r 副本一直沒有 新消息寫入的情況下follower 副本會(huì)一直發(fā)送拉取請(qǐng)求流礁，井且總收到空的拉取結(jié)果涕俗，這樣徒耗資源，顯然不太合理 神帅。
Kafka 選擇了延時(shí) 操作來(lái)處理這種情況咽袜。 Kafka 在處理拉取請(qǐng)求時(shí)，會(huì)先讀取一次日志文件 枕稀， 如果收集不到足夠多fetchMinBytes询刹，由參數(shù) fetch.mi口.bytes 配置，默認(rèn)值為 l)的消息萎坷， 那么就會(huì)創(chuàng)建一個(gè)延時(shí)拉取操作( DelayedFetch) 以等待拉取到足夠數(shù)量 的消息 凹联。當(dāng)延 時(shí)拉取操 作執(zhí)行時(shí)，會(huì)再讀取一次 日志文件哆档，然后將拉取結(jié)果返回給 follower 副本蔽挠。 延時(shí)拉取操作也會(huì) 有一個(gè)專門的延時(shí)操作管理器負(fù)責(zé)管理，大體的脈絡(luò)與延時(shí)生產(chǎn)操作相同瓜浸，不再贅述澳淑。 如果拉 取進(jìn)度一直沒有追趕上 leader副本，那么在拉取 leader副本的消息時(shí)一般拉取的消息大小都會(huì) 不小于 fetc出1inBytes插佛，這樣 Kafka也就不會(huì)創(chuàng)建相應(yīng)的延時(shí)拉取操作杠巡， 而是立即返回拉取結(jié)果。
延時(shí)拉取操作同樣是由超時(shí)觸發(fā)或外部事件觸發(fā)而被執(zhí)行的雇寇。 超時(shí)觸發(fā)很好理解氢拥，就是等 到超時(shí)時(shí)間之后觸發(fā)第 二次讀取 日志文件的操作 。外部事件觸發(fā)就稍復(fù)雜了一些锨侯，因?yàn)槔≌?qǐng) 求不單單 由 follower 副本發(fā)起 嫩海，也可以由消費(fèi)者客戶端發(fā)起，兩種情況所對(duì)應(yīng)的外部事件也是 不同的囚痴。如果是 follower 副本的延時(shí)拉取叁怪，它的外部事件就是消息追加到了 leader 副本的本地日志文件中 :如果是消費(fèi)者客戶端的延時(shí)拉取，它的外部事件可以簡(jiǎn)單地理解為 HW 的增長(zhǎng)深滚。
目前版本的 Kafka 壓引入了事務(wù)的概念奕谭，對(duì)于消費(fèi)者或 follower 副本而言 耳璧，其默認(rèn)的事務(wù) 隔離級(jí) 別為 “read_uncommitted” 。 不過(guò)消費(fèi)者可以通過(guò)客戶端參數(shù) isolation . level 將事 務(wù)隔離級(jí) 別設(shè)置為“ read_committed" (注意: follower 副本不可以將事務(wù)隔離級(jí)別修改為這個(gè) 值〉展箱，這樣消費(fèi)者拉取不到生產(chǎn)者已經(jīng)寫 入?yún)s尚未提交的消息 旨枯。 對(duì)應(yīng)的消費(fèi)者的延時(shí)拉取 ， 它 的外部事件實(shí)際上會(huì)切換為由LSO (LastStableOffset)的增長(zhǎng)來(lái)觸發(fā)混驰。 LSO是HW之前除去未 提交的事務(wù)消息的最大偏移量攀隔， LSO運(yùn)HW，
6.4 控制器
在 Kafka 集群中會(huì)有一個(gè)或多個(gè) broker栖榨，其中有一個(gè) broker 會(huì)被選舉為控制器( Kafka Controller)昆汹，它負(fù)責(zé)管理整個(gè)集群中所有分區(qū)和副本的狀態(tài)。當(dāng)某個(gè)分區(qū)的 leader 副本出現(xiàn)故 障時(shí)婴栽，由控制器負(fù)責(zé)為該分區(qū)選舉新的 leader副本满粗。當(dāng)檢測(cè)到某個(gè)分區(qū)的 ISR集合發(fā)生變化時(shí)， 由控制器負(fù)責(zé)通知所有 broker更新其元數(shù)據(jù)信息愚争。當(dāng)使用 kafka-topics.sh 腳本為某個(gè) topic 增加分區(qū)數(shù)量時(shí)映皆，同樣還是由控制器負(fù)責(zé)分區(qū)的重新分配 。
6.4.1 控制器的選舉及異澈渲Γ恢復(fù)
Kafka 中的控制器選舉工作依賴于 ZooKeeper捅彻，成功競(jìng)選為控制器的 broker會(huì)在 ZooKeeper中創(chuàng)建/ controller 這個(gè)臨時(shí)( EPHEMERAL)節(jié)點(diǎn)，此臨時(shí)節(jié)點(diǎn)的內(nèi)容參考如下 :
{ ” version ” : 1 鞍陨，” brokerid ”: 0 , ”timestamp” · ” 1 5 2 9 2 1 0 2 7 8 9 8 8 ” }
其中version在目前版本中固定為1, broker工d表示成為控制器的broker的id編號(hào)步淹， tim e stamp 表示競(jìng)選成為控制器時(shí)的時(shí)間戳。
在任意時(shí)刻诚撵，集群中有且僅有一個(gè)控制器缭裆。每個(gè) broker 啟動(dòng)的時(shí)候會(huì)去嘗試讀取 /controller 節(jié)點(diǎn)的 brokerid 的值，如果讀取到 brokerid 的值不為一l寿烟，則表示己經(jīng)有其 他 broker 節(jié) 點(diǎn)成功競(jìng)選為控制器澈驼，所以當(dāng)前 broker 就會(huì)放棄競(jìng)選;如果 ZooKeeper 中不存在 /controller 節(jié)點(diǎn)，或者這個(gè)節(jié)點(diǎn)中的數(shù)據(jù)異常韧衣，那么就會(huì)嘗試去創(chuàng)建/ controller 節(jié)點(diǎn)盅藻。 當(dāng)前 broker 去創(chuàng)建節(jié)點(diǎn)的時(shí)候，也有可能其他 broker 同時(shí)去嘗試創(chuàng)建這個(gè)節(jié)點(diǎn)畅铭，只有創(chuàng)建成功 的那個(gè) broker 才會(huì)成為控制 器，而創(chuàng)建失敗的 broker 競(jìng)選失敗 勃蜘。 每個(gè) broker 都會(huì)在內(nèi)存中保存 當(dāng)前控制器的 brokerid 值硕噩，這個(gè)值可以標(biāo)識(shí)為 activeControllerld。
ZooKeeper 中還有一個(gè)與控制器有關(guān)的/ controller_epoch 節(jié)點(diǎn)缭贡，這個(gè)節(jié)點(diǎn)是持久 (PERSISTENT)節(jié)點(diǎn)炉擅，節(jié)點(diǎn)中存放的是一個(gè)整型的 controller epoch 值辉懒。 controller
epoch 用于記錄控制器發(fā)生變更的次數(shù)，即記錄當(dāng)前的控制器是第幾代控制器谍失，我們也可以稱 之為“控制器的紀(jì)元”眶俩。
controller epoch 的初始值為 l，即集群中第一個(gè)控制器的紀(jì)元為 l快鱼，當(dāng)控制器發(fā)生變更 時(shí)颠印，每選出一個(gè)新的控制器就將該字段值加 1。每個(gè)和控制器交互的請(qǐng)求都會(huì)攜帶 controller epoch 這個(gè)宇段抹竹，如果請(qǐng)求的 controller_epoch 值小于內(nèi)存中的 controller_epoch值线罕， 則認(rèn)為這個(gè)請(qǐng)求是向己經(jīng)過(guò)期的控制器所發(fā)送的請(qǐng)求，那么這個(gè)請(qǐng)求會(huì)被認(rèn)定為無(wú)效的請(qǐng)求窃判。 如果請(qǐng)求的 controller epoch 值大于內(nèi)存中的 controller_epoch 值钞楼，那么說(shuō)明 己經(jīng)有 新的控制器當(dāng)選了 。 由此可見袄琳， Kafka 通過(guò) controller epoch 來(lái)保證控制器的唯一性询件，進(jìn)而保證相 關(guān)操作 的一致性。
具備控制器身份的broker需要比其他普通的broker多一份職責(zé)唆樊， 具體細(xì)節(jié)如下:
- 監(jiān)聽分區(qū)相關(guān)的變化雳殊。為 ZooKeeper 中的/admin/reassign partitions 節(jié)點(diǎn)注 冊(cè) PartitionReassignmentHandler， 用 來(lái) 處 理分區(qū)重分 配的 動(dòng) 作 窗轩。 為 ZooKeeper 中的 /工sr_change_not工f工cat工on 節(jié)點(diǎn)注冊(cè) IsrChangeNotificetionHandler夯秃，用來(lái)處理 ISR 集合變更 的動(dòng)作 。 為 ZooKeeper 中的 /admin/preferred-replica-election 節(jié) 點(diǎn)添加 PreferredReplicaElectionHandler痢艺，用來(lái)處理優(yōu)先副本 的選舉動(dòng)作仓洼。
- 監(jiān)聽 主題 相 關(guān) 的 變 化 。為 ZooKeeper 中的 /brokers/topics 節(jié) 點(diǎn)添 加 TopicChangeHandl町堤舒， 用來(lái) 處 理主題增減 的 變 化: 為 ZooKeeper 中 的 /admin/ de l e t e topics 節(jié)點(diǎn)添加 TopicDeletionHandler色建，用來(lái)處理刪 除主題 的動(dòng)作。
- 監(jiān)聽 broker相關(guān)的變化舌缤。為 ZooKeeper中的/brokers/ids 節(jié)點(diǎn)添加 BrokerChangeHandler, 用來(lái)處理 broker增減的變化箕戳。
- 從 ZooKeeper 中讀取獲取當(dāng)前所有與主題、分區(qū)及 broker 有關(guān)的信息并進(jìn)行相應(yīng)的管 理国撵。 對(duì)所 有主題 對(duì) 應(yīng) 的 ZooKeeper 中的 /brokers/topics/<topic>節(jié) 點(diǎn)添 加 PartitionModificationsHandler陵吸， 用來(lái)監(jiān)聽主題中的分區(qū)分配變化 。
- 啟動(dòng)并管理分區(qū)狀態(tài)機(jī)和副本狀態(tài)機(jī)介牙。
- 如果參數(shù) auto.leader.rebalance.enable 設(shè)置為 true壮虫，則還會(huì)開啟一個(gè)名為 “auto-leader-rebalance-task” 的定時(shí)任務(wù)來(lái)負(fù)責(zé)維護(hù)分區(qū)的優(yōu)先副本的均衡。
控制器在選舉成功之后會(huì)讀取 ZooKeeper 中各個(gè)節(jié)點(diǎn)的數(shù)據(jù)來(lái)初始化上下文信息 (ControllerContext)，并且需要管理這些上下文信息囚似。 比如為某個(gè)主題增加了若干分區(qū) 剩拢， 控制 器在負(fù)責(zé)創(chuàng)建這些分區(qū)的同 時(shí)要更新上下文信息 ， 并且需要將這些變更信息 同步到其他普通的 broker 節(jié)點(diǎn)中饶唤。不管是監(jiān)聽器觸發(fā)的事件徐伐，還是定時(shí)任務(wù)觸發(fā)的事件，或者是其他事件( 比如 ControlledShutdown募狂， 具體可以參考 6.4.2 節(jié))都會(huì)讀取或更新控制器中的上下文信息办素， 那么這 樣就會(huì)涉及多線程間的同步 。 如果單純使用鎖機(jī)制來(lái)實(shí)現(xiàn) 熬尺， 那么整體的性能會(huì)大打折扣 摸屠。針對(duì) 這一現(xiàn)象， Kafka 的控制器使用單線程基于事件隊(duì)列的模型粱哼， 將每個(gè)事件都做一層封裝季二， 然后 按照事 件 發(fā)生 的 先后順序暫存 到 LinkedB!ockingQueue 中 ，最后使 用 一個(gè)專 用的 線程 (ControllerEventThread)按照 FIFO (FirstInputFirstOutput揭措，先入先出)的原則順序序處理各個(gè)
事件胯舷，這樣不需要鎖機(jī)制就可以在多線程間維護(hù)線程安全， 具體可以參考圖 6-140
在 Kafka 的早期版本中绊含，并沒有采用 Kafka Controler 這樣一個(gè)概念來(lái)對(duì)分區(qū)和副本的狀態(tài) 進(jìn)行管理桑嘶，而是依賴于 ZooKeeper， 每個(gè) broker都會(huì)在 ZooKeeper上為分區(qū)和副本注冊(cè)大量的 監(jiān)昕器( Watcher) 躬充。當(dāng) 分區(qū)或副本狀態(tài)變化 時(shí) 逃顶，會(huì)喚醒很多不必要的監(jiān)昕器，這種嚴(yán)重依賴ZooKeeper 的設(shè)計(jì)會(huì)有腦裂充甚、羊群效應(yīng) 以政，以及造成 ZooKeeper 過(guò)載的隱患( 舊版的消費(fèi)者客戶 端存在同樣的問題， 詳 細(xì)內(nèi) 容參考 7.2.1 節(jié)) 伴找。 在目前的新版本的設(shè)計(jì)中盈蛮，只有 Kafka Controller 在 ZooKeeper 上注冊(cè)相應(yīng)的監(jiān)昕器，其 他的 broker 極少需要再監(jiān) 聽 ZooKeeper 中的 數(shù)據(jù)變化 技矮， 這樣省去了很多不必要的麻煩抖誉。不過(guò)每個(gè) broker還是會(huì)對(duì)/controller 節(jié)點(diǎn)添加監(jiān)聽器， 以 此來(lái)監(jiān) 昕此節(jié)點(diǎn)的 數(shù)據(jù)變化 (ControllerCbangeHandler) 衰倦。
當(dāng)/controller 節(jié)點(diǎn)的數(shù)據(jù)發(fā)生變化時(shí)袒炉， 每個(gè) broker 都會(huì)更新自身內(nèi)存中保存的 activeControllerld。 如果 broker 在數(shù)據(jù)變更前是控制器耿币，在數(shù)據(jù)變更后自身的 brokerid 值與 新的 activeControllerld 值不一致梳杏，那么就需要“退位” ， 關(guān)閉相應(yīng)的資源淹接，比如關(guān)閉狀態(tài)機(jī)十性、 注銷相應(yīng)的監(jiān)聽器等 。 有可能控制器由于異常而下線塑悼，造成/ controller 這個(gè)臨時(shí)節(jié)點(diǎn)被自 動(dòng)刪除 ; 也有可能是其他原因?qū)⒋斯?jié)點(diǎn)刪除了 劲适。
當(dāng)/controller 節(jié)點(diǎn)被刪除時(shí)，每個(gè) broker都會(huì)進(jìn)行選舉厢蒜，如果 broker在節(jié)點(diǎn)被刪除前 是控制器霞势，那么在選舉前還需要有 一個(gè)“退位”的動(dòng)作 。 如果有特殊需要 斑鸦，則可以手 動(dòng)刪除 /controller 節(jié)點(diǎn)來(lái)觸發(fā)新 一輪的選舉 愕贡。 當(dāng)然關(guān) 閉控制器所對(duì)應(yīng) 的 broker，以 及手動(dòng) 向 /controller 節(jié)點(diǎn)寫入新的 brokerid 的所對(duì)應(yīng)的數(shù)據(jù)巷屿，同樣可 以觸發(fā)新一輪的選舉 固以。
6.4.2 優(yōu)雅關(guān)閉
如何優(yōu)雅地關(guān)閉 Kafka?筆者在做測(cè)試的時(shí)候經(jīng)常性使用 jps (或者 ps ax)配合 kill -9 的方式來(lái)快速 關(guān)閉 Kafka broker 的服務(wù)進(jìn)程，顯然 kill -9 這種 “強(qiáng)殺”的方式并不夠優(yōu)雅嘱巾， 它并不會(huì)等待 Kafka 進(jìn)程合理關(guān)閉一些資源及保存一些運(yùn)行數(shù)據(jù)之后再實(shí)施關(guān)閉動(dòng)作憨琳。在有些 場(chǎng)景中，用戶希望主動(dòng)關(guān)閉正常運(yùn)行的服務(wù)旬昭，比如更換硬件篙螟、操作系統(tǒng)升級(jí)、修改 Kafka 配置 等问拘。如果依然使用上述方式關(guān)閉就略顯粗暴 遍略。
那么合理的操作應(yīng)該是什么呢? Kafka 自身提供了 一 個(gè)腳本工具，就是存放在其 bin 目錄 下的 kafka-server-stop . sh骤坐，這個(gè)腳本的內(nèi)容非常簡(jiǎn)單绪杏，具體內(nèi)容如下:
PIDS=♀(ps ax I grep -i ’kafka\.Kafka’ I grep java I grep -v grep I awk ’(print $1)’)
if [ -z "♀PIDS” ] ; then
echo ”No kafka server to stop” exit 1
else
kill -s TERM ♀PIDS fi
可以看出 kafka-server stop.sh 首先通過(guò) ps ax 的方式找出正在運(yùn)行 Kafka 的進(jìn)程 號(hào) PIDS，然后使用 kill -s TERM $PIDS 的方式來(lái)關(guān)閉或油。 但是這個(gè)腳本在很多時(shí)候并不奏 效寞忿，這一點(diǎn)與ps命令有關(guān)系。 在Linux操作系統(tǒng)中顶岸， ps命令限制輸出的字符數(shù)不得超過(guò)頁(yè)大 小 PAGE_SIZE腔彰， 一般 CPU 的內(nèi)存管理單元(Memory Management Unit，簡(jiǎn)稱 MMU)的 PAGE_SIZE 為 4096辖佣。 也就是說(shuō)霹抛， ps 命令的輸出的字符串長(zhǎng)度限制在 4096 內(nèi)，這會(huì)有什么問 題呢?我們使用 ps ax 命 令來(lái)輸出與 Kafka 進(jìn)程相 關(guān)的信息卷谈，如圖 6-15 所示 杯拐。
細(xì)心的讀者可以留 意到 白色部分中的信息并沒有打印全，因?yàn)榧航?jīng)達(dá)到了 4096 的字符數(shù)的 限制。 而且打印的信息里面也沒有 kafka-server-stop.sh 中 ps ax I grep -i ’ kafka \ . Kafka ’所需要 的“ kafka.Kafka端逼，朗兵，這個(gè)關(guān)鍵字段，因?yàn)檫@個(gè)關(guān)鍵字段在 4096 個(gè)字 符的范圍之外顶滩。與 Kafka 進(jìn)程有關(guān)的輸出信息太長(zhǎng)余掖，所以 kafka-server-stop . sh 腳本在很
多情況 下并不 會(huì)奏效。
注意要點(diǎn):Kafak服務(wù)啟動(dòng)的入口叫Kafka.Kafka scala語(yǔ)言寫的object
那么怎么解決這種問題呢?我們先來(lái)看一下 ps 命令的相關(guān)源碼(Linux 2.6.x 源碼的/fs/proc/base.c 文件中的部分 內(nèi)容):
我們可以看到 ps 的輸出長(zhǎng)度 len 被硬編碼成小于等于 PAGE SIZE 的大小礁鲁，那么我們調(diào) 大這個(gè) PAGE SIZE 的大小不就可以了嗎?這樣是肯定行不通的盐欺，因?yàn)閷?duì)于一個(gè) CPU來(lái)說(shuō)，它 的 MMU 的頁(yè)大小 PAGE SIZE 的值是固定的仅醇，無(wú)法通過(guò)參數(shù)調(diào)節(jié)冗美。 要想改變 PAGE SIZE 的 大小，就必須更換成相應(yīng)的 CPU析二，顯然這也太過(guò)于“興師動(dòng)眾”了 粉洼。還有一種辦法是 ，將上面 代碼中的 PAGE SIZE 換成一個(gè)更大的其他值甲抖，然后 重新編譯漆改，這個(gè)辦法對(duì)于大多數(shù)人來(lái)說(shuō)不 太適用， 需要掌握一定深度的Linux的相關(guān)知識(shí)准谚。
那么 有沒有 其他的辦法呢?這里我們可以 直接修改 kafka-server-stop.sh 腳本的內(nèi) 容挫剑，將其中的第一行命 令修改 如下:
PIDS=$(ps ax I grep -i ’kafka’ I grep java I grep -v grep I awk ’ {print $1)’)
即把“\ .Kafka”去掉，這樣在絕大多數(shù)情況下是可以奏效的柱衔。如果有極端情況樊破，即使這 樣 也不能 關(guān) 閉，那么只 需要按 照以下兩個(gè)步驟就可以優(yōu)雅地關(guān)閉 Kafka 的服務(wù)進(jìn)程:
(1 )獲取 Kafka 的服務(wù)進(jìn)程號(hào) PIDS唆铐。 可以使用 Java 中的 jps 命令或使用 Linux 系統(tǒng)中 的 ps 命令來(lái)查看哲戚。
(2)使用kill -s TERM ♀PIDS或kill 15 ♀PIDS的方式來(lái)關(guān)閉進(jìn)程，注意千萬(wàn) 不要使用 kill 斗 的方式艾岂。
為什么這樣關(guān)閉的方式會(huì)是優(yōu)雅的? Kafka 服務(wù)入口程序中有一個(gè)名為“ kafka-shutdown- hock”的關(guān)閉鉤子 顺少， 待 Kafka 進(jìn)程捕獲終止信號(hào)的時(shí)候會(huì)執(zhí)行這個(gè)關(guān)閉鉤子中的內(nèi)容，其中除 了正常關(guān)閉一些必要的資源王浴，還會(huì)執(zhí)行一 個(gè) 控制關(guān)閉( ControlledShutdown)的 動(dòng) 作 脆炎。 使用 ControlledShutdown的方式關(guān)閉 Kafka有兩個(gè)優(yōu)點(diǎn): 一是可以讓消息完全同步到磁盤上，在服務(wù) 下次重新上線時(shí)不需要進(jìn)行日志的恢復(fù)操作 ; 二是 ControllerShutdown 在關(guān) 閉服務(wù)之前氓辣，會(huì)對(duì) 其上的 leader 副本進(jìn)行遷移秒裕，這樣就可以減少分區(qū)的不可用時(shí)間 。
若要成功執(zhí)行 Co由olledShutdown 動(dòng)作還需要有一個(gè)先決條件钞啸， 就是參數(shù) controlled. shutdown.enable 的值需要設(shè)置為 true几蜻，不過(guò)這個(gè)參數(shù)的默認(rèn)值就為 true喇潘，即默認(rèn)開始此 項(xiàng)功能 。 ControlledShutdown 動(dòng)作如果執(zhí)行不成功還會(huì)重試執(zhí)行梭稚，這個(gè)重試的動(dòng)作由參數(shù) controlled.shutdown.max.retries 配置颖低，默認(rèn)為 3 次， 每次重試的間隔由參數(shù) controlled . shutdown . retry .backoff .ms 設(shè)置哨毁，默認(rèn)為 5000ms
下面我們具體探討 ControlledShutdown 的整個(gè)執(zhí)行過(guò)程枫甲。
參考圖 ι16源武， 假設(shè)此時(shí)有兩個(gè) broker扼褪，其中待 關(guān)閉的 brok町 的 id 為 x, Kafka 控制器所對(duì) 應(yīng) 的 broker 的 id 為 y。待關(guān) 閉的 broker 在執(zhí)行 ControlledShutdown 動(dòng) 作時(shí) 首先與 Kafka 控 制器 建立專用連接(對(duì)應(yīng)圖 6-16 中的步驟1) 粱栖， 然后發(fā)送 ControlledShutdownRequest 請(qǐng)求话浇， ControlledShutdownRequest 請(qǐng)求中只有一個(gè) brokerld 字段， 這個(gè) brokerld 字段的值設(shè)置為自身 的brokerId的值闹究，即x (對(duì)應(yīng)圖6-16中的步驟2) 幔崖。
Kafka 控制 器在收到 ControlledShutdownRequest 請(qǐng)求之后會(huì)將與待關(guān) 閉 broker 有關(guān)聯(lián) 的所 有分區(qū)進(jìn)行專 門 的處理，這里的“有關(guān)聯(lián)”是指分區(qū)中有副本位于這個(gè)待 關(guān) 閉的 broker 之上 (這 里會(huì)涉及 Kafka控制器與待關(guān)閉 broker之間的多次交互動(dòng)作渣淤，涉及 leader副本的遷移和副本的 關(guān)閉動(dòng)作赏寇，對(duì)應(yīng)圖 6-16 中的步驟3〉。
如果這些分區(qū)的副本數(shù)大于 1 且 leader副本位于待關(guān)閉 broker上价认，那么需要實(shí)施 leader副 本的遷移及新的 ISR 的 變更嗅定。具體的選舉分配的方案由專用的選舉器 ControlledShutdown- LeaderSelector提供
如果這些分區(qū)的副本數(shù)只是大于 1, leader 副本并不位于待關(guān)閉 broker 上，那么就由 Kafka 控制器來(lái)指導(dǎo)這些副本的 關(guān)閉 用踩。 如果這些分區(qū)的副本數(shù)只是為 1渠退， 那么這個(gè)副本的關(guān)閉動(dòng)作會(huì) 在整個(gè) ControlledShutdown 動(dòng)作執(zhí)行之后由副本管理器來(lái)具體實(shí)施 。
對(duì)于分區(qū)的副本數(shù)大于 l 且 leader 副本位于待關(guān)閉 broker 上的這種情況脐彩，如果在 Kafka 控 制器處理之后 leader 副本還沒有成功遷移碎乃，那么會(huì)將這些沒有成功遷移 leader 副本的分區(qū)記錄 下來(lái)，并且寫入 ControlledShutdownResponse 的響應(yīng)(對(duì)應(yīng)圖 6-16 中的步驟4惠奸，整個(gè)ControlledShutdown 動(dòng)作是 一個(gè)同步阻塞的過(guò)程) 梅誓。ControlledShutdownResponse 的結(jié)構(gòu)如圖 6-18 所示。
待關(guān)閉的 broker 在收到 ControlledShutdownResponse 響應(yīng)之后佛南，需要判斷整個(gè) Con位olledShu創(chuàng)own 動(dòng)作是否執(zhí)行成功梗掰，以此來(lái)進(jìn)行可能的 重試或繼續(xù) 執(zhí)行接下來(lái)的關(guān)閉 資源 的動(dòng)作 。 執(zhí)行成功的 標(biāo)準(zhǔn)是 Con位olledShutdownResponse 中 error_code 字段值為 0共虑，并且 partitions remaining 數(shù)組字段為空愧怜。
在了解了整個(gè) ControlledShutdown 動(dòng)作的具體細(xì)節(jié)之后，我們不難看出這一切實(shí)質(zhì)上都是 由 ControlledShutdownRequest請(qǐng)求引發(fā)的妈拌，我們完全可以自己開發(fā)一個(gè)程序來(lái)連接 Kafka控制 器拥坛，以此來(lái)模擬對(duì)某個(gè) broker 實(shí)施 ControlledShutdown 的動(dòng)作蓬蝶。為了實(shí)現(xiàn)方便，我們可以對(duì) KafkaAdminC!ient 做 一些擴(kuò)展來(lái)達(dá)到目的猜惋。
6.4.3 分區(qū) leader 的選舉
分區(qū) leader副本的選舉由控制器負(fù)責(zé)具體實(shí)施丸氛。當(dāng)創(chuàng)建分區(qū)(創(chuàng)建主題或增加分區(qū)都有創(chuàng) 建分區(qū)的動(dòng)作〉或分區(qū)上線(比如分區(qū)中原先的 leader 副本下線，此時(shí)分區(qū)需要選舉一個(gè)新的 leader 上 線來(lái)對(duì)外提供服務(wù))的時(shí)候都需要執(zhí)行 leader 的選舉動(dòng)作著摔，對(duì)應(yīng)的選舉策略為 OftlinePartitionLeaderElectionStrategy灭将。 這種策略的基本思路是按照 AR 集合中副本的順序查找 第一個(gè)存活的副本，并且這個(gè)副本在 JSR集合中脊岳。 一個(gè)分區(qū)的 AR集合在分配的時(shí)候就被指定包个， 并且只要不發(fā)生重分配的情況，集合內(nèi)部副本的順序是保持不變的摹察，而分區(qū)的 ISR 集合中副本 的順序可能會(huì)改變 恩掷。
注意這里是根據(jù)AR的順序而不是ISR的順序進(jìn)行選舉的。舉個(gè)例子供嚎， 集群中有3個(gè)節(jié)點(diǎn): brokerO黄娘、 brokerl 和 broker2， 在某一時(shí)刻具有 3個(gè)分區(qū)且副本因子為 3 的主題 topic扣ader的具 體信息如下 :
如 果 ISR 集合中 沒有可用的副本 克滴， 那么此時(shí)還要再檢查一下所配置的 unclean .leader . e l e c t i o n .四 able 參數(shù)(默認(rèn)值為 false) 逼争。 如果這個(gè)參數(shù)配置為 true，那么表示允許從非 ISR 列表中 的選舉 leader劝赔，從 AR 列表中找到 第一個(gè)存活的副本 即為 leader誓焦。
當(dāng)分區(qū)進(jìn)行重分配(可以先回顧一下 4.3.2節(jié)的內(nèi)容)的時(shí)候也需要執(zhí)行 leader的選舉動(dòng)作，對(duì)應(yīng)的選舉策略為 ReassignPartiti望忆。此eaderElectionStrategy罩阵。這個(gè)選舉策略 的思路 比較簡(jiǎn)單 : 從
重分配的 AR 列表中找到第 一個(gè)存活的副本，且這個(gè)副本在目前的 ISR 列表 中 启摄。
還有 一 種情況會(huì)發(fā)生 leader 的選舉稿壁，當(dāng)某節(jié)點(diǎn)被優(yōu)雅地關(guān) 閉 ( 也 就是 執(zhí) 行 ControlledShutdown)時(shí)，位于這個(gè)節(jié)點(diǎn)上的 lead巳r副本都會(huì)下線歉备，所以與此對(duì)應(yīng)的分區(qū)需要執(zhí) 行 leader 的選舉傅是。與此對(duì)應(yīng) 的選舉策略( ControlledShutdownPartitionLeaderElectionStrategy)為 : 從 AR 列表中找到第一個(gè)存活的副本，且這個(gè)副本在目前的 ISR列表中蕾羊，與此同時(shí)還要確保這 個(gè)副本不處于正在被關(guān)閉的節(jié)點(diǎn)上 喧笔。
6.5 參數(shù)解密
如果 broker端沒有顯式配置 listeners (或 advertised. listeners)使用 IP地址， 那么最好將 bootstrap.server 配置成主機(jī)名而不要使用 IP 地址龟再，因?yàn)?Kafka 內(nèi)部使用的是 全稱域名(FullyQualifiedDomainName) 书闸。 如果不統(tǒng)一， 則會(huì)出現(xiàn)無(wú)法獲取元數(shù)據(jù)的異常利凑。
6.5.1 broker.id
broker . id 是 broker 在啟動(dòng)之前必須設(shè)定 的參數(shù)之一浆劲，在 Kafka 集群 中 嫌术，每個(gè) broker 都 有唯一的 id (也可以記作 brokerld)值用來(lái)區(qū)分彼此。 broker 在啟動(dòng)時(shí)會(huì)在 ZooKeeper 中的 /brokers/ids 路徑下創(chuàng)建一個(gè)以當(dāng)前 brokerId為名稱的虛節(jié)點(diǎn)牌借， broker 的健康狀態(tài)檢查就依 賴于此虛節(jié)點(diǎn)度气。當(dāng) broker 下線時(shí)，該虛節(jié)點(diǎn)會(huì)自動(dòng)刪除膨报，其他 broker 節(jié)點(diǎn)或客戶端通過(guò)判斷 /brokers/ids 路徑下是否有此 broker 的 brokerld 節(jié)點(diǎn)來(lái)確定該 broker 的健康狀態(tài)磷籍。
可以通過(guò) broker 端的配置文件 config/server.properties 里的 broker . id 參數(shù)來(lái)配置 brokerid， 默認(rèn)情況下broker.id值為 l现柠。在Kafka中院领， brokerld值必須大于等于0才有可能 正常啟動(dòng)，但這里并不是只能通過(guò)配置文件 config/server.properties 來(lái)設(shè)定這個(gè)值晒旅，還可以通過(guò) meta.properties 文件或 自動(dòng)生成功能來(lái)實(shí)現(xiàn)栅盲。
首先了解一下 meta.properties 文件， meta.properties 文件中的內(nèi)容參考 如下:
#Sun May 27 23:03:04 CST 2018
version=O
broker.id=O
meta.properties文件中記錄了與當(dāng)前 Kafka版本對(duì)應(yīng)的一個(gè) version字段废恋，不過(guò)目前只有一個(gè)為0的固定值。還有一個(gè)broker.id扒寄，即brokerid值鱼鼓。 broker在成功啟動(dòng)之后在每個(gè)日志根
目錄下都會(huì)有一個(gè) meta.properties 文件 。
m巳ta.properties 文件與 broker . id 的關(guān)聯(lián)如下 :
Cl)如果 log.d工r 或 log.d工rs 中配置了多個(gè)日志根目錄该编，這些日志根目錄中的 meta.properties 文件所配置的 broker . id 不一致則會(huì)拋出 InconsistentBrokerldException 的 異常迄本。
(2)如果 config/server.pr叩erties配置文件里配置的 broker.工d的值和 meta.properties文 件里的 broker . 工d 值不 一致 ，那么同樣會(huì)拋出 InconsistentBrokerldException 的 異常 课竣。
( 3 )如 果 config/server.properties 配置文件中井未配置 broker .工d 的值嘉赎，那么就以 meta.properties文件中的 broker. id值為準(zhǔn)。
(4)如果沒有 meta.properties文件于樟，那么在獲取合適的 broker. id值之后會(huì)創(chuàng)建一個(gè)新 的 meta.prop巳rties文件并將 broker.id值存入其中公条。
如果 config/server.properties 配置文件中并未配置 broker. id，并且日志根目錄中也沒有 任何 meta.properties 文件( 比如第-次啟動(dòng) 時(shí) ) 迂曲，那么應(yīng)該如何 處理呢 ?
Kafka 還提供 了另外兩個(gè) broker 端參數(shù) : broker. id.generatio口.enable 和 reserved. broker.max.id來(lái)配合生成新的 brokerId靶橱。broker. id.geηeratio口.enable 參數(shù)用來(lái)配置是否開啟自動(dòng)生成 brokerId 的功能，默認(rèn)情況下為廿ue路捧， 即開啟此功能 关霸。自 動(dòng)生 成的 brokerId 有一個(gè)基準(zhǔn)值，即自動(dòng)生成的 brokerId 必 須超過(guò)這個(gè)基準(zhǔn)值杰扫，這個(gè)基準(zhǔn)值通過(guò) reserverd .broker.max . id 參數(shù)配置队寇，默認(rèn)值為 1000。 也就是說(shuō)章姓，默認(rèn)情況下自動(dòng)生成的 brokerId 從 1001 開始 佳遣。
自動(dòng)生成的 brokerId的原理是先往 ZooKeeper中的/brokers/seqid節(jié)點(diǎn)中寫入一個(gè)空宇 符串 炭序，然后獲取返回的 Stat 信 息中 的 version 值 ，進(jìn)而將 version 的值和 reserved.broker .max . id 參數(shù)配置 的值相加苍日。先往節(jié)點(diǎn)中 寫入數(shù)據(jù)再獲取 Stat 信息 惭聂， 這樣 可以確保返回的 version 值大于 0 ，進(jìn)而 就可 以確 保生 成的 brokerId 值大于 reserved.broker.max.id 參數(shù) 配置的值相恃，符合非自動(dòng)生成的 broker .id 的 值在 [O, reserved.broker.max.id]區(qū)間設(shè)定辜纲。
初始化時(shí) ZooKeeper 中 /brokers/seq工d 節(jié) 點(diǎn) 的狀態(tài)如下 :
[zk: xxx.xxx.xxx.xxx:2181/kafka(CONNECTED) 6] get /brokers/seqid null
cZxid = Ox200001b2b
ctime =Mon Nov 13 17:39:54 CST 2018
mZx 工d = Ox20000lb2b
mtime = Mon Nov 13 17: 39 :54 CST 2018 pZxid = Ox20000lb2b
cversion = 0
dataV ersion = 0
aclV ersion = 0 ephemeralOwner = OxO dataLength = O numChildren = 0
可以看到 dataVersion=O，這個(gè)就是前面所說(shuō)的version,在插入一個(gè)空字符串之后拦耐，dataVersio就自增1表示數(shù)據(jù)發(fā)生了變更耕腾， 這樣通過(guò) ZooKeeper 的這個(gè)功能來(lái)實(shí)現(xiàn)集群層面的序號(hào)遞增，整體上相當(dāng)于一個(gè)發(fā)號(hào)器 杀糯。
[zk: xxx.xxx.xxx.xxx:2181/kafka(CONNECTED) 7] set /brokers/seqid ”” cZxid = Ox200001b2b
ctime =Mon Nov 13 17:39:54 CST 2017
mZxid = Ox2000e6eb2
mtime = Mo口 May 28 18:19 : 03 CST 2018 pZxid = Ox200001b2b
cversion = 0
dataV ersion = 1
aclV ersion = 0 ephemeralOwner = OxO dataLength = 2 numChildren = 0
大多數(shù)情況下我們一般通過(guò)井且習(xí)慣于用最普通的 config/server.properties 配置文件的方式 來(lái)設(shè)定 brokerld 的值扫俺，如果知曉其中的細(xì)枝末節(jié)，那么在遇到諸如 InconsistentBrokerldException 異常時(shí)就可以處理得游刃有余固翰，也可以通過(guò)自動(dòng)生成 brokerId 的功能來(lái)實(shí)現(xiàn)一些另類的功能 狼纬。
6.5.2 bootstrap.servers
bootstrap.servers 不僅是 Kafka Producer、 Kafka Consumer 客戶端 中的必備 參數(shù) 骂际，而 且在 KafkaConnect疗琉、 KafkaStreams和 KafkaAdminClient中都有涉及， 是一個(gè)至關(guān)重要的參數(shù)歉铝。
如果你使用過(guò)舊版的生產(chǎn)者或舊版的消費(fèi)者客戶端盈简，那么你可能還會(huì)對(duì) bootstrap . servers 相關(guān)的另外兩個(gè)參數(shù) metada .broker .list 和 zookeeper.connect 有些許印象，這3個(gè)參數(shù)也見證了 Kafka 的升級(jí)變遷太示。
我們一般可以簡(jiǎn)單地認(rèn)為 bootstrap.servers 這個(gè)參數(shù)所要指定的就是將要連接的Kafka集群的 broker地址列表柠贤。不過(guò)從深層次的意義上來(lái)講，這個(gè)參數(shù)配置的是用來(lái)發(fā)現(xiàn) Kafka 集群元數(shù)據(jù)信息的服務(wù)地址类缤。為了更加形象地說(shuō)明問題臼勉，我們先來(lái)看一下圖 6-19。
客戶端 KafkaProducer1 與 Kafka Cluster 直連呀非，這是客戶端給我們的既定印象坚俗，而事實(shí)上客戶端連接 Kafka集群要經(jīng)歷以下3個(gè)程，如圖 6-19 中的右邊所示岸裙。
Cl)客戶端 KafkaProducer2 與 bootstrap.servers 參數(shù)所指定的 Server連接猖败，井發(fā)送MetadataRequest 請(qǐng)求來(lái)獲取集群的元數(shù)據(jù)信息 。
(2) Server在收到 MetadataRequest請(qǐng)求之后降允，返回 MetadataResponse給 KafkaProducer2,在 MetadataResponse 中包含了集群的元數(shù)據(jù)信息恩闻。
(3)客戶端 KafkaProducer2 收到的 MetadataResponse 之后解析出其中包含的集群元數(shù)據(jù)信息，然后與集群中的各個(gè)節(jié)點(diǎn)建立連接剧董，之后就可以發(fā)送消息了幢尚。
在絕大多數(shù)情況下破停， Kafka 本身就扮演著第一步和第二步中的 Server 角色，我們完全可以 將這個(gè) Server 的角色從 Kafka 中剝離出來(lái)尉剩。我們可以在這個(gè) Server 的角色上大做文章真慢，比如添 加一些路由的功能、負(fù)載均衡的功能 理茎。
下面演示如何將 Server 的角色與 Kafka 分開黑界。默認(rèn)情況下，客戶端從 Kafka 中的 某個(gè)節(jié)點(diǎn) 來(lái)拉取集群的元數(shù)據(jù)信息皂林，我們可以將所拉取的元數(shù)據(jù)信息復(fù)制一份存放到 Server 中朗鸠，然后對(duì) 外提供這份副本的內(nèi)容信 息。
由此可見础倍，我們首先需要做的就是獲取集群信息的副本烛占，可以在 Kafka 的 org.apache.kafka. comrnon.request.M巳tadataResponse 的構(gòu)造函數(shù)中嵌入代碼來(lái)復(fù)制信息， MetadataResponse 的構(gòu)造 函數(shù)如下所示沟启。
public MetadataResponse(int throttleTimeMs , List<Node> brokers , St ring clusterid忆家， 工nt controllerid,
L工st<Top工cMetadata> topicMetadata) { this.throttleTimeMs = throttleTimeMs;
this.brokers =brokers ;
this .controller= getControllerNode(controllerid, brokers) ; this.topicMetadata = topicMetadata;
this.clusterid = clusterid;
//客戶端在獲取集群的元數(shù)據(jù)之后會(huì)調(diào)用 這個(gè)構(gòu)造函數(shù)，所以在這里嵌入代碼將 5 個(gè)成
//員變量的值保存起來(lái)美浦，為后面的 Server 提供 內(nèi)容
獲取集群元數(shù)據(jù)的副本之后弦赖，我們就可以實(shí)現(xiàn)一個(gè)服務(wù)程序來(lái)接收 MetadataRequest請(qǐng)求和 MetadataResponse，從零開始構(gòu)建 一個(gè)這樣的服務(wù)程序也需要不少的工作量 浦辨， 需要實(shí)現(xiàn)對(duì) MetadataRequest與 MetadataResponse相關(guān)協(xié)議解析和包裝，這里不妨再修改一下 Kafka 的代碼沼沈，返回讓其只提供 Server相關(guān)的內(nèi)容流酬。整個(gè)示例的架構(gòu)如圖 6-20所示。
為了演示方便列另， 圖 6-20 中的 Kafka Clusterl 和 Kafka Cluster2都只包含一個(gè) broker節(jié)點(diǎn)芽腾。 Kafka Clusterl 扮演的是 Se凹er 的角色，下面我們修改它的代碼讓其返回 Kafka Cluster2 的集群 元數(shù)據(jù)信息 页衙。假設(shè)我們己經(jīng)通過(guò)前面一步的操作獲取了 Kafka Cluster2 的集群元數(shù)據(jù)信息摊滔，在 Kafka Clusterl 中將這份副本回放。
修改完 Kafka ClusterI 的代碼之后我們將它和 Kafka Cluster2 都啟動(dòng)起來(lái)店乐，然后創(chuàng)建一個(gè)生 產(chǎn)者 KafkaProducer 來(lái)持續(xù)發(fā)送消息艰躺，這個(gè) KafkaProducer 中的 bootstrap . servers 參數(shù)配 置為 Kafka Cluster!的服務(wù)地址。 我們?cè)賱?chuàng)建一個(gè)消費(fèi)者 KafkaConsumer 來(lái)持續(xù)消費(fèi)消息眨八，這 個(gè) KafkaConsumer 中的 bootstrap . servers 參數(shù)配置為 Kafka Cluster2 的服務(wù)地址 腺兴。
實(shí)驗(yàn)證明 ， KatkaP1oducer 中發(fā)送的消息都流入 Kafka Cluster2 并被 KafkaConsumer 消費(fèi)廉侧。 查看 Kafka Cluster1 中的日志文件页响， 發(fā)現(xiàn)并沒有消息流入篓足。 如果此時(shí)我們?cè)訇P(guān)閉 Kafka Cluster1 的服務(wù)，會(huì)發(fā)現(xiàn) KafkaProducer和 KafkaConsumer都運(yùn)行完好闰蚕，已經(jīng)完全沒有 KafkaCluster! 的 任何事情了 栈拖。
這里只是為了講解 bootstrap.servers 參數(shù)所代表的真正含義而做的一些示例演示， 筆者并不建議在真實(shí)應(yīng)用中像示例中的一樣分離 出 Server 的角色 没陡。
在舊版的生產(chǎn)者客戶端(Scala 版本)中還沒有 bootstrap . servers 這個(gè)參數(shù) 涩哟， 與此對(duì) 應(yīng)的是 metadata.broker. list參數(shù)。metadata.broker. list這個(gè)參數(shù)很直觀诗鸭，metadata 表示元數(shù)據(jù)染簇， broker.list表示 broker的地址列表， 從取名我們可以看出這個(gè)參數(shù)很直接地表示所 要連接 的 Kafka broker 的地址强岸，以此 獲取元數(shù)據(jù)锻弓。 而 新版 的 生產(chǎn)者客戶端 中的bootstrap.servers 參數(shù)的取名顯然更有內(nèi)涵，可以直觀地翻譯為“引導(dǎo)程序的服務(wù)地址” 蝌箍， 這樣在取名上就多了一層 “代理”的空間青灼，讓人可以遐想出 Server角色與 Ka僅a分離的可能。 在舊版的消費(fèi)者客戶端( Scala版本)中也沒有 bootstrap. servers 這個(gè)參數(shù)妓盲，與此對(duì)應(yīng)的 是 zookeeper . connect 參數(shù)杂拨，意為通過(guò) ZooKeeper 來(lái)建立消費(fèi)連接。
很多讀者從 0.8.x 版本開始沿用到現(xiàn)在的 2.0.0 版本悯衬， 對(duì)于版本變遷的客戶端中出現(xiàn)的bootstrap.servers弹沽、 metadata.broker.list、 zookeeper.connect 參數(shù)往往不是 很清楚筋粗。這一現(xiàn)象還存在 Kafka 所提供的諸多腳本之中策橘，在這些腳本中連接 Kafka 采用的選項(xiàng) 參數(shù)有 一 bootstrap-server、 --broker-list 和一一zookeeper (分別與前面的 3 個(gè)參數(shù) 對(duì)應(yīng)〕娜亿，這讓很 多 Kafka 的老手也很難分 辨哪個(gè)腳本該用哪個(gè)選項(xiàng)參數(shù)丽已。
--bootstrap-server 是一個(gè)逐漸盛行的選項(xiàng)參數(shù)，這一點(diǎn)毋庸置疑买决。而一broker-list 己經(jīng)被淘汰沛婴，但在 2.0.0 版本中還沒有完全被摒棄，在 kafka-console-producer.sh腳本中還是使用 的這個(gè)選項(xiàng)參數(shù)督赤，在后 續(xù)的 Kafka 版本中 可能會(huì)被替代為 --bootstrap-server 嘁灯。 一 zookeeper 這個(gè)邊項(xiàng)參數(shù)也逐漸被替代，在目前的 2.0.0 版本中够挂， kafka-console-consumer.sh 中已經(jīng)完全沒有了它的 影子旁仿，但并不意味著這個(gè)參數(shù)在其 他腳 本中 也被摒棄了 。在 kafka-topics.sh腳本 中還是使用的一 zookeeper 這個(gè)選項(xiàng)參數(shù)，并且在未來(lái)的可期版本中也不 見得會(huì)被替換枯冈，因?yàn)?kafka-topics.sh腳本實(shí)際上操縱的就是 ZooKeeper 中的節(jié)點(diǎn)毅贮， 而不是 Kafka 本身，它并沒有被替代的必要尘奏。
