今天我們來(lái)深度剖析下「Kafka Broker 端網(wǎng)絡(luò)架構(gòu)和請(qǐng)求處理流程」是如何設(shè)計(jì)的? 相信使用過(guò) Kafka 的朋友都知道其吞吐量可以高達(dá)百萬(wàn)，但很少人理解其中的設(shè)計(jì)原理。
那么 Kafka Broker 端網(wǎng)絡(luò)架構(gòu)和請(qǐng)求處理到底是使用了哪些高大上的技術(shù)？它到底解決了什么問(wèn)題？究竟是怎么解決的？
只有了解了這些， 我們才能深刻掌握 Kafka 服務(wù)端設(shè)計(jì)精髓所在，更加深刻理解一個(gè)高并發(fā)、高性能服務(wù)端架構(gòu)該如何設(shè)計(jì)。
認(rèn)真讀完這篇文章，我相信你會(huì)對(duì)Kafka Broker請(qǐng)求處理流程和網(wǎng)絡(luò)架構(gòu)設(shè)計(jì)實(shí)現(xiàn)細(xì)節(jié)，有更加深刻的理解。
這篇文章干貨很多，希望你可以耐心讀完。
01 總體概述
要想理解 Kafka Broker 請(qǐng)求處理架構(gòu)設(shè)計(jì)，我們需要從簡(jiǎn)單請(qǐng)求處理模型來(lái)說(shuō)起。
對(duì)于日常系統(tǒng)開(kāi)發(fā)，我們都知道是基于 Request/Response 的模式來(lái)實(shí)現(xiàn)的， 對(duì)于 Kafka 來(lái)說(shuō)， 無(wú)論是 Producer 端、Consumer 端 還是 Broker 端，他們之間的請(qǐng)求交互也都是基于「Request/Response」模式來(lái)完成的。比如，客戶端會(huì)通過(guò)網(wǎng)絡(luò)發(fā)送消息生產(chǎn)請(qǐng)求給 Broker，而 Broker 處理完成后，會(huì)發(fā)送對(duì)應(yīng)的響應(yīng)給到客戶端。
下面，我會(huì)從自我設(shè)計(jì)角度出發(fā)，如果是我們會(huì)如何設(shè)計(jì)，帶你一步步演化出來(lái)「kafka Broker 的網(wǎng)絡(luò)請(qǐng)求處理」架構(gòu)。
在這個(gè)過(guò)程中，你會(huì)看到 Kafka 在處理請(qǐng)求的過(guò)程中會(huì)遇到哪些高性能和高并發(fā)問(wèn)題，以及架構(gòu)為什么要這樣演進(jìn)，從而理解 Kafka 這么設(shè)計(jì)的意義和精妙之處。
02 順序處理模式
我們從最簡(jiǎn)單的網(wǎng)絡(luò)編程思路處理方式講起。
因?yàn)閷?duì)于 Kafka Broker 來(lái)說(shuō)就是用來(lái)接收生產(chǎn)者發(fā)送過(guò)來(lái)的請(qǐng)求，那這個(gè)時(shí)候最簡(jiǎn)單的實(shí)現(xiàn)大概是這樣的：
如上述代碼所示：我們可以理解 Kafka 每個(gè)服務(wù)器啟動(dòng)起來(lái)后就是一個(gè) while 循環(huán)， 不斷的 accept 生產(chǎn)者提交上來(lái)的請(qǐng)求， 然后進(jìn)行處理并存儲(chǔ)到磁盤(pán)上，這種方式實(shí)現(xiàn)最簡(jiǎn)單，也非常好理解，但是這種方式存在2個(gè)致命的缺陷？
1）請(qǐng)求****阻塞：只能順序處理每個(gè)請(qǐng)求，即每個(gè)請(qǐng)求都必須等待前一個(gè)請(qǐng)求處理完畢才能得到處理。
2）吞吐量非常差：由于只能順序處理，無(wú)法并發(fā)，效率太低，所以吞吐量非常差，只適合請(qǐng)求發(fā)送非常不頻繁的系統(tǒng)。
從上面來(lái)看很明顯，如果你的 Kafka 系統(tǒng)請(qǐng)求并發(fā)量很大，意味著要處理的時(shí)間就會(huì)越久。那按照前面我們提到的 Kafka「吞吐量」的標(biāo)準(zhǔn)，這個(gè)方案遠(yuǎn)遠(yuǎn)無(wú)法滿足我們對(duì)高性能、高并發(fā)的要求。
那有什么更好的方案可以快速處理請(qǐng)求嗎？
接下來(lái)我們可以試著采取這個(gè)方案：獨(dú)立線程異步處理模式。
03 多線程異步處理模式
既然同步方式會(huì)阻塞請(qǐng)求，吞吐量差， 我們可以嘗試著使用獨(dú)立線程異步方式進(jìn)行處理， 即經(jīng)典的 connection per thread 模型， 那這個(gè)時(shí)候的實(shí)現(xiàn)大概是這樣的：
如上述代碼所示：同上還是一個(gè) while 循環(huán)不斷的 accept 生產(chǎn)者提交上來(lái)的請(qǐng)求，但是這時(shí)候 Kafka 系統(tǒng)會(huì)為每個(gè)請(qǐng)求都創(chuàng)建一個(gè)「單獨(dú)的線程」來(lái)處理。
這個(gè)實(shí)現(xiàn)方案的好處就是：
1）吞吐量****稍強(qiáng)：相對(duì)上面同步方式的方案，一定程度上極大地提高了服務(wù)器的吞吐量。
2）非阻塞：它是完全異步的，每個(gè)請(qǐng)求的處理都不會(huì)阻塞下一個(gè)請(qǐng)求。
但同樣缺陷也同樣很明顯：即為每個(gè)請(qǐng)求都創(chuàng)建線程的做法開(kāi)銷(xiāo)很大，在某些高并發(fā)場(chǎng)景下會(huì)壓垮整個(gè)服務(wù)?？梢?jiàn)，這個(gè)方案也只適用于請(qǐng)求發(fā)送頻率很低的業(yè)務(wù)場(chǎng)景。還是無(wú)法滿足我們對(duì)高性能、高并發(fā)的要求。
既然這種方案還是不能滿足， 那么我們究竟該使用什么方案來(lái)支撐高并發(fā)呢？
這個(gè)時(shí)候我們可以想想我們?nèi)粘ｉ_(kāi)發(fā)用到的7層負(fù)載Nginx或者Redis在處理高并發(fā)請(qǐng)求的時(shí)候是使用什么方案呢？
從上面啟發(fā)你可以看出，提升系統(tǒng) I/O 并發(fā)性能的關(guān)鍵思路就是：事件驅(qū)動(dòng)。
想必大家已經(jīng)猜到了，沒(méi)錯(cuò)，就是「多路復(fù)用」。那么Kafka 是不是也是采用這種方案來(lái)實(shí)現(xiàn)呢？
這里我們先考慮采用基于「事件驅(qū)動(dòng)」的設(shè)計(jì)方案，當(dāng)有事件觸發(fā)時(shí)，才會(huì)調(diào)用處理器進(jìn)行數(shù)據(jù)處理。
04 Reactor 模式
在高性能網(wǎng)絡(luò)編程領(lǐng)域，有一個(gè)非常著名的模式——Reactor模式。那么何為「Reactor模式」，首先它是基于事件驅(qū)動(dòng)的，有一個(gè)或多個(gè)并發(fā)輸入源，有一個(gè)Service Handler，有多個(gè)Request Handler；這個(gè)Service Handler會(huì)同步的將輸入的請(qǐng)求輪詢地分發(fā)給相應(yīng)的Request Handler進(jìn)行處理。
借助于 Doug Lea（就是那位讓人無(wú)限景仰的大爺）的 "Scalable IO in Java" 中講述的Reactor模式。
"Scalable IO in Java" 的地址是：
http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
簡(jiǎn)單來(lái)說(shuō)，Reactor 模式特別適合應(yīng)用于處理多個(gè)客戶端并發(fā)向服務(wù)器端發(fā)送請(qǐng)求的場(chǎng)景。這里借用大神 PDF 中的一幅圖來(lái)說(shuō)明 Reactor 架構(gòu)：
從上面這張圖中，我們可以看出多個(gè)客戶端會(huì)發(fā)送請(qǐng)求給到 Reactor。Reactor 有個(gè)請(qǐng)求分發(fā)線程 Dispatcher，也就是圖中的綠色的 Acceptor，它會(huì)將不同的請(qǐng)求下分發(fā)到多個(gè)工作線程中處理。
在這個(gè)架構(gòu)中，Acceptor 線程只是用來(lái)進(jìn)行請(qǐng)求分發(fā)，所以非常輕量級(jí)，因此會(huì)有很高的吞吐量。而這些工作線程可以根據(jù)實(shí)際系統(tǒng)負(fù)載情況動(dòng)態(tài)調(diào)節(jié)系統(tǒng)負(fù)載能力，從而達(dá)到請(qǐng)求處理的平衡性。
基于上面的 Reactor 架構(gòu)， 我們來(lái)看看如果是我們?cè)撊绾卧O(shè)計(jì) Kafka 服務(wù)端的架構(gòu)？
1）這里我們采用多路復(fù)用方案，Reactor 設(shè)計(jì)模式，并引用 Java NIO 的方式可以更好的解決上面并發(fā)請(qǐng)求問(wèn)題。
2）當(dāng) Client 端將請(qǐng)求發(fā)送到 Server 端的時(shí)候， 首先在 Server 端有個(gè)多路復(fù)用器（Selector），然后會(huì)啟動(dòng)一個(gè) Accepter 線程將 OP_CONNECT 事件注冊(cè)到多路復(fù)用器上， 主要用來(lái)監(jiān)聽(tīng)連接事件到來(lái)。
3）當(dāng)監(jiān)聽(tīng)到連接事件后，就會(huì)在多路復(fù)用器上注冊(cè) OP_READ 事件， 這樣 Cient 端發(fā)送過(guò)來(lái)的請(qǐng)求， 都會(huì)被接收到。如果請(qǐng)求特別多的話， 我們這里進(jìn)行優(yōu)化， 創(chuàng)建一個(gè) Read HandlePool 線程池。
4）當(dāng) Read HandlePool 線程池接收到請(qǐng)求數(shù)據(jù)后，最終會(huì)交給 Handler ThreadPool 線程池進(jìn)行后續(xù)處理。比如如果是生產(chǎn)者發(fā)送過(guò)來(lái)的請(qǐng)求，肯定會(huì)解析請(qǐng)求體，處理并最終存儲(chǔ)到磁盤(pán)中，待處理完后要返回處理結(jié)果狀態(tài)， 這時(shí)候就由它在多路復(fù)用器上注冊(cè) OP_WRITE 事件來(lái)完成。這樣多路復(fù)用器遍歷到OP_WRITE 事件后就會(huì)將請(qǐng)求返回到 Client 端。
5）在上圖中我們看到在整個(gè)流程中還有一個(gè) MessageQueue 的隊(duì)列組件存在， 為什么要加這個(gè)組件呢？ 我們可以想象一下， 如果請(qǐng)求量非常大，直接交給 Handler ThreadPool 線程池進(jìn)行處理， 可能會(huì)出現(xiàn)該線程池處理不過(guò)來(lái)的情況發(fā)生，如果處理不過(guò)來(lái)，也會(huì)出現(xiàn)阻塞瓶頸。所以這里我們?cè)?Server 端內(nèi)部也設(shè)計(jì)一個(gè)消息隊(duì)列， 起到一個(gè)緩沖的作用，Handler ThreadPool 線程池會(huì)根據(jù)自己的負(fù)載能力進(jìn)行處理。
以上就是我們引用了「多路復(fù)用」的設(shè)計(jì)方案，但是 Kafka Broker 端就是這樣的架構(gòu)設(shè)計(jì)方案嗎？如果我們是 Kafka 系統(tǒng)架構(gòu)的設(shè)計(jì)者，采用這樣的架構(gòu)設(shè)計(jì)方案會(huì)不會(huì)還是有什么問(wèn)題，有沒(méi)有哪個(gè)環(huán)節(jié)會(huì)出現(xiàn)系統(tǒng)性能瓶頸呢？
這是個(gè)值得思考的問(wèn)題， 很考驗(yàn)?zāi)愕募軜?gòu)設(shè)計(jì)能力。
細(xì)心的讀者可能會(huì)發(fā)現(xiàn)：對(duì)于 Kafka 這種超高并發(fā)系統(tǒng)來(lái)說(shuō)，一個(gè) Selector 多路復(fù)用器是 Hold 不住的，從上圖可以得出，我們監(jiān)聽(tīng)這些連接、接收請(qǐng)求、處理響應(yīng)結(jié)果都是同一個(gè) Selector 在進(jìn)行處理，很容易成為系統(tǒng)性能瓶頸。
接下來(lái)，我們將進(jìn)一步進(jìn)行優(yōu)化，為了減輕當(dāng)前 Selector 的處理負(fù)擔(dān)，引入另外一個(gè)Selector 處理隊(duì)列，如下圖所示：
1）首先上圖是目前我認(rèn)為最接近 Kafka Broker 真實(shí)架構(gòu)設(shè)計(jì)方案的。
2）整體架構(gòu)跟上一版的類似，只不過(guò)這里多引入了一個(gè)多 Selector 處理隊(duì)列，原來(lái)的 Selector 只負(fù)責(zé)監(jiān)聽(tīng)連接， 這時(shí)候有讀者就會(huì)有疑問(wèn)，請(qǐng)求量超級(jí)大的時(shí)候，一個(gè) Selector 會(huì)不會(huì)成為瓶頸呢？ 這里可以大可放心， 這時(shí)候它的工作非常單一，是完全能 hold 住的。
3）那么對(duì)于我們接收請(qǐng)求、處理請(qǐng)求、返回狀態(tài)操作都會(huì)交由多 Selector 處理隊(duì)列，至于這里到底需要多少個(gè) Selector，又會(huì)跟什么參數(shù)和配置有關(guān)系，我們后續(xù)再進(jìn)行分析，總之這里記住有多個(gè) Selector 就行了，這樣系統(tǒng)壓力就會(huì)被分散處理。
4）另外我們要搞清楚的一點(diǎn)就是對(duì)于 Kafka 服務(wù)端指的是每個(gè) Broker 節(jié)點(diǎn)，如果我們的服務(wù)集群總共有10個(gè)節(jié)點(diǎn)， 每個(gè)節(jié)點(diǎn)內(nèi)部都是上面的這樣的架構(gòu)，這樣我們就有理由相信如果采用這樣的架構(gòu)設(shè)計(jì)方案，是可以支持高并發(fā)和高性能的。
架構(gòu)設(shè)計(jì)方案演進(jìn)到這里，基本上已經(jīng)差不多了，接下來(lái)我們看看 Kafka 真實(shí)超高并發(fā)的網(wǎng)絡(luò)架構(gòu)是如何設(shè)計(jì)的。
05 Kafka 超高并發(fā)網(wǎng)絡(luò)架構(gòu)
在上面 Kafka 高性能、高吞吐量架構(gòu)演進(jìn)的時(shí)候，我們提到了 Java NIO 以及 Reactor 設(shè)計(jì)模式。實(shí)際上，搞透了「Kafka 究竟是怎么使用 NIO 來(lái)實(shí)現(xiàn)網(wǎng)絡(luò)通信的」，不僅能讓我們掌握 Kafka 請(qǐng)求處理全流程處理，也能讓我們對(duì) Reactor 設(shè)計(jì)模式有更深的理解，還能幫助我們解決很多實(shí)際問(wèn)題。
在[<kafka 三高架構(gòu)設(shè)計(jì)剖析>] 和 [<聊聊 Kafka Producer 那點(diǎn)事>] 中也講過(guò)關(guān)于 Kafka 網(wǎng)絡(luò)架構(gòu)，讀者可以去查閱，今天我們?cè)俅紊钊氲膶?duì)其進(jìn)行剖析。
那么接下來(lái)我們就來(lái)深入剖析下 Kafka 的 NIO 通訊機(jī)制吧。
我們先從整體上看一下完整的網(wǎng)絡(luò)通信層架構(gòu)，如下圖所示：
1）從上圖中我們可以看出，Kafka 網(wǎng)絡(luò)通信架構(gòu)中用到的組件主要由兩大部分構(gòu)成：SocketServer 和 RequestHandlerPool。
2）SocketServer 組件是 Kafka 超高并發(fā)網(wǎng)絡(luò)通信層中最重要的子模塊。它包含 Acceptor 線程、Processor 線程和 RequestChannel 等對(duì)象，都是網(wǎng)絡(luò)通信的重要組成部分。它主要實(shí)現(xiàn)了 Reactor 設(shè)計(jì)模式，主要用來(lái)處理外部多個(gè) Clients（這里的 Clients 可能包含 Producer、Consumer 或其他 Broker）的并發(fā)請(qǐng)求，并負(fù)責(zé)將處理結(jié)果封裝進(jìn) Response 中，返還給 Clients。
3）RequestHandlerPool 組件就是我們常說(shuō)的 I/O 工作線程池，里面定義了若干個(gè) I/O 線程，主要用來(lái)執(zhí)行真實(shí)的請(qǐng)求處理邏輯。
4）這里注意的是：跟 RequestHandler 相比， 上面所說(shuō)的Acceptor、Processor 線程 還有 RequestChannel 等都不做請(qǐng)求處理， 它們只是請(qǐng)求和響應(yīng)的「搬運(yùn)工」。
接下來(lái)我們來(lái)具體聊聊SocketServer中的實(shí)現(xiàn)原理，這里先來(lái)講講：
- Acceptor 線程
- Processor 線程
以Kafka 2.5版本，源碼位置：https://github.com/apache/kafka/blob/2.5.0-rc3/core/src/main/scala/kafka/network/SocketServer.scala
1、我們先來(lái)聊聊 ****Acceptor 線程
在經(jīng)典的 Reactor 設(shè)計(jì)模式有個(gè) 「Dispatcher」 的角色，主要用來(lái)接收外部請(qǐng)求并分發(fā)給下面的實(shí)際處理線程。通過(guò)上面分析我們知道在 Kafka 網(wǎng)絡(luò)架構(gòu)設(shè)計(jì)中，這個(gè) Dispatcher 就是「Acceptor 線程」。
Acceptor 線程是用來(lái)接收和創(chuàng)建外部 TCP 連接的線程。在Broker 端每個(gè) SocketServer 實(shí)例只會(huì)創(chuàng)建一個(gè) Acceptor 線程。它的主要功能就是創(chuàng)建連接，并將接收到的 Request 請(qǐng)求傳遞給下游的 Processor 線程處理。
/**
這里重點(diǎn)看下 Acceptor 線程中三個(gè)非常關(guān)鍵且重要的屬性和方法：
1.1、nioSelector：它就是我們所熟悉的 Java NIO 庫(kù)中的 Selector 對(duì)象實(shí)例，所有網(wǎng)絡(luò)通信組件實(shí)現(xiàn) Java NIO 機(jī)制的基礎(chǔ)。
1.2、processors：通過(guò)源碼我們可以知道在Acceptor 線程在初始化時(shí)，需要?jiǎng)?chuàng)建對(duì)應(yīng)的 Processor 線程池。由此可以得出，Processor 線程是在 Acceptor 線程中管理和維護(hù)的。
1.3、run方法：它是處理 Reactor 模式中分發(fā)邏輯的主要實(shí)現(xiàn)方法。
1）從上述源碼中，我們可以看出 Acceptor 線程主要使用了 Java NIO 的 Selector 以及 SocketChannel 的方式循環(huán)的輪詢準(zhǔn)備就緒的 I/O 事件。
2)這里的 I/O 事件主要是指網(wǎng)絡(luò)連接創(chuàng)建事件即：SelectionKey.OP_ACCEPT。
3）這樣注冊(cè)好事件后，一旦后續(xù)接收到連接請(qǐng)求后，Acceptor 線程就會(huì)指定一個(gè) Processor 線程，并將該請(qǐng)求交給它并創(chuàng)建網(wǎng)絡(luò)連接用于后續(xù)處理。
2、接下來(lái)我們來(lái)聊聊 ****Processor 線程：
從上面分析我們知道 Acceptor 只是做了請(qǐng)求入口連接處理的，那么，真正創(chuàng)建網(wǎng)絡(luò)連接以及分發(fā)網(wǎng)絡(luò)請(qǐng)求是由 Processor 線程來(lái)完成的。
override def run(): Unit = {
從上面 Processor 線程源碼，可以看出 Kafka 的代碼邏輯實(shí)現(xiàn)的非常好，各個(gè)子方法的邊界非常清楚。
這里我們就不展開(kāi)源碼分析了， 更深入詳細(xì)的等到源碼分析專題再進(jìn)行。我們簡(jiǎn)單的看下 Processor 線程初始化時(shí)要做的事情。
看上面代碼最后部分，我們知道每個(gè) Processor 線程在創(chuàng)建時(shí)都會(huì)創(chuàng)建 3 個(gè)隊(duì)列。
1）newConnections 隊(duì)列: 它主要是用來(lái)保存要?jiǎng)?chuàng)建的新連接信息，也就是SocketChannel 對(duì)象，目前是硬編碼隊(duì)列長(zhǎng)度大小為20。每當(dāng) Processor 線程接收到新的連接請(qǐng)求時(shí)，都會(huì)將對(duì)應(yīng)的 SocketChannel 對(duì)象放入隊(duì)列，等到后面創(chuàng)建連接時(shí)，從該隊(duì)列中獲取 SocketChannel，然后注冊(cè)新的連接。
2）inflightResponse 隊(duì)列：它是一個(gè)臨時(shí)的 Response 隊(duì)列， 當(dāng) Processor 線程將 Repsonse 返回給 Client 之后，要將 Response 放入該隊(duì)列。它存在的意義：由于有些 Response 回調(diào)邏輯要在 Response 被發(fā)送回 Request 發(fā)送方后，才能執(zhí)行，因此需要暫存到臨時(shí)隊(duì)列。
3）ResponseQueue 隊(duì)列：它主要是存放需要返回給Request 發(fā)送方的所有 Response 對(duì)象。通過(guò)源碼得知：每個(gè) Processor 線程都會(huì)維護(hù)自己的 Response 隊(duì)列。
06 請(qǐng)求處理核心流程剖析
上面深入的剖析了 Kafka 超高并發(fā)網(wǎng)絡(luò)架構(gòu) 以及 SocketServer 中的 Acceptor 線程跟 Processor 線程的實(shí)現(xiàn)原理， 接下來(lái)我們來(lái)將請(qǐng)求處理核心流程給串起來(lái)。
只有搞透這部分的實(shí)現(xiàn)原理，才能幫助我們有針對(duì)性的進(jìn)行 Broker端請(qǐng)求處理的性能調(diào)優(yōu)。
比如：在上面網(wǎng)絡(luò)架構(gòu)圖，有兩個(gè)參數(shù)跟整個(gè)流程有關(guān)系，分別是num.network.threads、num.io.threads。如果我們不掌握請(qǐng)求處理的整個(gè)流程，就不能更好的對(duì)此進(jìn)行調(diào)整，來(lái)達(dá)到更高的性能要求。
其中 num.io.threads 就是 I/O 工作線程池的大小配置，即 KafkaRequestHandlerPool 線程池，它才是「真正處理 Kafka 請(qǐng)求」的地方。
源碼位置：
/**
下面我們結(jié)合 Kafka 超高并發(fā)網(wǎng)絡(luò)架構(gòu)圖來(lái)講解下一個(gè)完整請(qǐng)求處理核心流程：
1）Clients 發(fā)送請(qǐng)求給 Acceptor 線程。
2）Acceptor 線程會(huì)創(chuàng)建 NIO Selector 對(duì)象，并創(chuàng)建 ServerSocketChannel 實(shí)例，然后將Channel 和 OP_ACCEPT 事件到 Selector 多路復(fù)用器上。
3）Acceptor 線程還會(huì)默認(rèn)創(chuàng)建3個(gè)大小的 Processor 線程池，參數(shù)：num.network.threads, 并輪詢的將請(qǐng)求對(duì)象 SocketChannel 放入到連接隊(duì)列中（newConnections）。
4）這時(shí)候連接隊(duì)列就源源不斷有請(qǐng)求數(shù)據(jù)了，然后不停地執(zhí)行 NIO Poll, 獲取對(duì)應(yīng) SocketChannel 上已經(jīng)準(zhǔn)備就緒的 I/O 事件。
5）Processor 線程向 SocketChannel 注冊(cè)了 OP_READ/OP_WRITE 事件，這樣 客戶端發(fā)過(guò)來(lái)的請(qǐng)求就會(huì)被該 SocketChannel 對(duì)象獲取到，具體就是CompleteReceives。
6）這個(gè)時(shí)候客戶端就可以源源不斷進(jìn)行請(qǐng)求發(fā)送了，服務(wù)端通過(guò) Selector NIO Poll 不停的獲取準(zhǔn)備就緒的 I/O 事件。
7）然后根據(jù)Channel中獲取已經(jīng)完成的 Receive 對(duì)象，構(gòu)建 Request 對(duì)象，并將其存入到 Requestchannel 的 RequestQueue 請(qǐng)求隊(duì)列中 。
8）這個(gè)時(shí)候就該 I/O 線程池上場(chǎng)了，KafkaRequestHandler 線程循環(huán)地從請(qǐng)求隊(duì)列中獲取 Request 實(shí)例，然后交由KafkaApis 的 handle 方法，執(zhí)行真正的請(qǐng)求處理邏輯，并最終將數(shù)據(jù)存儲(chǔ)到磁盤(pán)中。
9）待處理完請(qǐng)求后，KafkaRequestHandler 線程會(huì)將 Response 對(duì)象放入 Processor 線程的 Response 隊(duì)列。
10）然后 Processor 線程通過(guò) Request 中的 ProcessorID 不停地從 Response 隊(duì)列中來(lái)定位并取出 Response 對(duì)象，返還給 Request 發(fā)送方。
至此，我們深入剖析Kafka請(qǐng)求處理「核心流程」。
07 系統(tǒng)調(diào)優(yōu)
搞透了 Kafka 超高并發(fā)網(wǎng)絡(luò)架構(gòu)設(shè)計(jì)和請(qǐng)求處理核心流程后，我們來(lái)聊聊 Broker 端參數(shù)調(diào)優(yōu)。
對(duì) Kafka 而言，性能一般是指吞吐量和延時(shí)。所以高吞吐量、低延時(shí)是我們調(diào)優(yōu) Kafka 集群的主要目標(biāo)。
Broker 端調(diào)優(yōu)主要就是合理地設(shè)置 Broker 端參數(shù)值，以匹配你的生產(chǎn)環(huán)境。另外還有一點(diǎn)要說(shuō)明的就是「保證服務(wù)器端和客戶端版本的一致」，做到這一點(diǎn)，就能獲得很多性能收益了。
num.network.threads
創(chuàng)建 Processor 處理網(wǎng)絡(luò)請(qǐng)求線程個(gè)數(shù)，建議設(shè)置為 Broker 當(dāng)前CPU核心數(shù)*2，這個(gè)值太低經(jīng)常出現(xiàn)網(wǎng)絡(luò)空閑太低而缺失副本。
num.io.threads
創(chuàng)建 KafkaRequestHandler 處理具體請(qǐng)求線程個(gè)數(shù)，建議設(shè)置為Broker磁盤(pán)個(gè)數(shù)*2。
num.replica.fetchers
建議設(shè)置為CPU核心數(shù)/4，適當(dāng)提高可以提升CPU利用率及 Follower同步 Leader 數(shù)據(jù)當(dāng)并行度
compression.type
建議采用lz4壓縮類型，壓縮可以提升CPU利用率同時(shí)可以減少網(wǎng)絡(luò)傳輸數(shù)據(jù)量。
compression.type!
建議采用lz4壓縮類型，壓縮可以提升CPU利用率同時(shí)可以減少網(wǎng)絡(luò)傳輸數(shù)據(jù)量。
log.flush.xxx
log.flush.scheduler.interval.ms
log.flush.interval.ms
log.flush.interval.messages
這幾個(gè)參數(shù)表示日志數(shù)據(jù)刷新到磁盤(pán)的策略，應(yīng)該保持默認(rèn)配置，刷盤(pán)策略讓操作系統(tǒng)去完成，由操作系統(tǒng)來(lái)決定什么時(shí)候把數(shù)據(jù)刷盤(pán)；如果設(shè)置來(lái)這個(gè)參數(shù)，可能對(duì)吞吐量影響非常大
auto.leader.rebalance.enable
表示是否開(kāi)啟leader自動(dòng)負(fù)載均衡，默認(rèn)true；我們應(yīng)該把這個(gè)參數(shù)設(shè)置為false，因?yàn)樽詣?dòng)負(fù)載均衡不可控，可能影響集群性能和穩(wěn)定。
08 總結(jié)
這里，我們一起來(lái)總結(jié)一下這篇文章的重點(diǎn)。
1、對(duì)于 Kafka 這樣一個(gè)優(yōu)秀的服務(wù)端系統(tǒng)架構(gòu)來(lái)說(shuō)，應(yīng)該遵循高可用、高性能、高并發(fā) 3 大原則。
2、本文從最簡(jiǎn)單的網(wǎng)絡(luò)編程思路出發(fā)一步一步演進(jìn)到 Reactor 設(shè)計(jì)模式，假設(shè)我們就是 Kafka 架構(gòu)的設(shè)計(jì)者，我們?cè)撊绾卧O(shè)計(jì)其服務(wù)端網(wǎng)絡(luò)架構(gòu)。
3、通過(guò)本文的深度剖析，提升系統(tǒng)I/O性能的核心是基于「事件驅(qū)動(dòng)」模型實(shí)現(xiàn)。
4、在剖析完服務(wù)端網(wǎng)絡(luò)架構(gòu)后，我們也深度剖析了 SocketServer中兩個(gè)最重要的線程：Acceptor 線程和 Processor 線程。
5、接著我們結(jié)合 Kafka 超高并發(fā)網(wǎng)絡(luò)架構(gòu)圖又梳理了 Kafka 請(qǐng)求處理核心流程。
6、最后給大家分析并做了 Broker 端系統(tǒng)調(diào)優(yōu)的方案。
