Kafka處理請求的流程是什么

本篇內(nèi)容主要講解“Kafka處理請求的流程是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Kafka處理請求的流程是什么”吧!

Reactor模式

在扯到Kafka之前我們先來說說Reactor模式，基本上只要是底層的高性能網(wǎng)絡(luò)通信就離不開Reactor模式。像Netty、Redis都是使用Reactor模式。

像我們以前剛學網(wǎng)絡(luò)編程的時候以下代碼可是非常的熟悉，新來一個請求，要么在當前線程直接處理了，要么新起一個線程處理。

在早期這樣的編程是沒問題的，但是隨著互聯(lián)網(wǎng)的快速發(fā)展，單線程處理不過來，也不能充分的利用計算機資源。

而每個請求都新起一個線程去處理，資源的要求就太高了，并且創(chuàng)建線程也是一個重操作。

說到這有人想到了，那搞個線程池不就完事了嘛，還要啥Reactor。

池化技術(shù)確實能緩解資源的問題，但是池子是有限的，池子里的一個線程不還是得候著某個連接，等待指示嘛。現(xiàn)在的互聯(lián)網(wǎng)時代早已突破C10K了。

因此引入的IO多路復用，由一個線程來監(jiān)視一堆連接，同步等待一個或多個IO事件的到來，然后將事件分發(fā)給對應(yīng)的Handler處理，這就叫Reactor模式。

網(wǎng)絡(luò)通信模型的發(fā)展如下 > 單線程 => 多線程 => 線程池 => Reactor模型

Kafka所采用的Reactor模型如下

Kafka Broker 網(wǎng)絡(luò)通信模型

簡單來說就是，Broker 中有個Acceptor(mainReactor)監(jiān)聽新連接的到來，與新連接建連之后輪詢選擇一個Processor(subReactor)管理這個連接。

而Processor會監(jiān)聽其管理的連接，當事件到達之后，讀取封裝成Request，并將Request放入共享請求隊列中。

然后IO線程池不斷的從該隊列中取出請求，執(zhí)行真正的處理。處理完之后將響應(yīng)發(fā)送到對應(yīng)的Processor的響應(yīng)隊列中，然后由Processor將Response返還給客戶端。

每個listener只有一個Acceptor線程，因為它只是作為新連接建連再分發(fā)，沒有過多的邏輯，很輕量，一個足矣。

Processor 在Kafka中稱之為網(wǎng)絡(luò)線程，默認網(wǎng)絡(luò)線程池有3個線程，對應(yīng)的參數(shù)是num.network.threads。并且可以根據(jù)實際的業(yè)務(wù)動態(tài)增減。

還有個 IO 線程池，即KafkaRequestHandlerPool，執(zhí)行真正的處理，對應(yīng)的參數(shù)是num.io.threads，默認值是 8。IO線程處理完之后會將Response放入對應(yīng)的Processor中，由Processor將響應(yīng)返還給客戶端。

可以看到網(wǎng)絡(luò)線程和IO線程之間利用的經(jīng)典的生產(chǎn)者 - 消費者模式，不論是用于處理Request的共享請求隊列，還是IO處理完返回的Response。

這樣的好處是什么？生產(chǎn)者和消費者之間解耦了，可以對生產(chǎn)者或者消費者做獨立的變更和擴展。并且可以平衡兩者的處理能力，例如消費不過來了，我多加些IO線程。

如果你看過其他中間件源碼，你會發(fā)現(xiàn)生產(chǎn)者-消費者模式真的是太常見了，所以面試題經(jīng)常會有手寫一波生產(chǎn)者-消費者。

源碼級別剖析網(wǎng)絡(luò)通信模型

Kafka 網(wǎng)絡(luò)通信組件主要由兩大部分構(gòu)成：SocketServer 和 KafkaRequestHandlerPool。

SocketServer

可以看出SocketServer旗下管理著，Acceptor 線程、Processor 線程和 RequestChannel 等對象。

data-plane和control-plane稍后再做分析，先看看RequestChannel是什么。

RequestChannel

關(guān)鍵的屬性和方法都已經(jīng)在下面代碼中注釋了，可以看出這個對象主要就是管理Processor和作為傳輸Request和Response的中轉(zhuǎn)站。

Acceptor

接下來我們再看看Acceptor

可以看到它繼承了AbstractServerThread，接下來再看看它run些啥

再來看看accept(key) 做了啥

很簡單，標準selector的處理，獲取準備就緒事件，調(diào)用serverSocketChannel.accept()得到socketChannel，將socketChannel交給通過輪詢選擇出來的Processor，之后由它來處理IO事件。 ##Processor 接下來我們再看看Processor，相對而言比Acceptor 復雜一些。

先來看看三個關(guān)鍵的成員

再來看看主要的處理邏輯。

可以看到Processor主要是將底層讀事件IO數(shù)據(jù)封裝成Request存入隊列中，然后將IO線程塞入的Response，返還給客戶端，并處理Response 的回調(diào)邏輯。

#KafkaRequestHandlerPool

IO線程池，實際處理請求的線程。

再來看看IO線程都干了些啥

很簡單，核心就是不斷的從requestChannel拿請求，然后調(diào)用handle處理請求。

handle方法是位于KafkaApis類中，可以理解為通過switch，根據(jù)請求頭里面不同的apikey調(diào)用不同的handle來處理請求。

我們再舉例看下較為簡單的處理LIST_OFFSETS的過程，即handleListOffsetRequest，來完成一個請求的閉環(huán)。

我用紅色箭頭標示了調(diào)用鏈。表明處理完請求之后是塞給對應(yīng)的Processor的。

最后再來個更詳細的總覽圖，把源碼分析到的類基本上都對應(yīng)的加上去了。

請求處理優(yōu)先級

上面提到的data-plane和control-plane是時候揭開面紗了。這兩個對應(yīng)的就是數(shù)據(jù)類請求和控制類請求。

為什么需要分兩類請求呢？直接在請求里面用key標明請求是要讀寫數(shù)據(jù)啊還是更新元數(shù)據(jù)不就行了嗎？

簡單點的說比如我們想刪除某個topic，我們肯定是想這個topic馬上被刪除的，而此時producer還一直往這個topic寫數(shù)據(jù)，那這個情況可能是我們的刪除請求排在第N個...等前面的寫入請求處理好了才輪到刪除的請求。實際上前面哪些往這個topic寫入的請求都是沒用的，平白的消耗資源。

再或者說進行Preferred Leader選舉時候，producer將ack設(shè)置為all時候，老leader還在等著follower寫完數(shù)據(jù)向他報告呢，誰知follower已經(jīng)成為了新leader，而通知它leader已經(jīng)變更的請求由于被一堆數(shù)據(jù)類型請求堵著呢，老leader就傻傻的在等著，直到超時。

就是為了解決這種情況，社區(qū)將請求分為兩類。

那如何讓控制類的請求優(yōu)先被處理？優(yōu)先隊列？

社區(qū)采取的是兩套Listener，即數(shù)據(jù)類型一個listener，控制類一個listener。

對應(yīng)的就是我們上面講的網(wǎng)絡(luò)通信模型，在kafka中有兩套！ kafka通過兩套監(jiān)聽變相的實現(xiàn)了請求優(yōu)先級，畢竟數(shù)據(jù)類型請求肯定很多，控制類肯定少，這樣看來控制類肯定比大部分數(shù)據(jù)類型先被處理！

迂回戰(zhàn)術(shù)啊。

控制類的和數(shù)據(jù)類區(qū)別就在于，就一個Porcessor線程，并且請求隊列寫死的長度為20。

到此，相信大家對“Kafka處理請求的流程是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進入相關(guān)頻道進行查詢，關(guān)注我們，繼續(xù)學習！