前言

經(jīng)過上篇文章的簡單實(shí)戰(zhàn)之后，今天來聊聊生產(chǎn)者將消息從客戶端發(fā)送到 Broker 上背后發(fā)生了哪些故事，看不看由你，但是我保證可以本篇文章你一定可以學(xué)到應(yīng)用背后的一些實(shí)質(zhì)東西。

本文我們從以下 4 個方面來探討下一條消息如何被準(zhǔn)確的發(fā)送到 Broker 的 partition 上。

1. 客戶端組件

2. 客戶端緩存存儲模型

3. 確定消息的 partition 位置

4. 發(fā)送線程的工作原理

客戶端組件

• KafkaProducer:

KafkaProducer 是一個生產(chǎn)者客戶端的進(jìn)程，通過該對象啟動生產(chǎn)者來發(fā)送消息。

• RecordAccumulator:

RecordAccumulator 是一個記錄收集器，用于收集客戶端發(fā)送的消息，并將收集到的消息暫存到客戶端緩存中。

• Sender:

Sender 是一個發(fā)送線程，負(fù)責(zé)讀取記錄收集器中緩存的批量消息，經(jīng)過一些中間轉(zhuǎn)換操作，將要發(fā)送的數(shù)據(jù)準(zhǔn)備好，然后交由 Selector 進(jìn)行網(wǎng)絡(luò)傳輸。

• Selector:

Selector 是一個選擇器，用于處理網(wǎng)絡(luò)連接和讀寫處理，使用網(wǎng)絡(luò)連接處理客戶端上的網(wǎng)絡(luò)請求。

通過使用以上四大組件即可完成客戶端消息的發(fā)送工作。消息在網(wǎng)絡(luò)中傳輸?shù)姆绞街荒芡ㄟ^二級制的方式，所以首先需要將消息序列化為二進(jìn)制形式緩存在客戶端，kafka 使用了雙端隊(duì)列的方式將消息緩存起來，然后使用發(fā)送線程（Sender）讀取隊(duì)列中的消息交給 Selector 進(jìn)行網(wǎng)絡(luò)傳輸發(fā)送給服務(wù)端（Broker）

以上為發(fā)送消息的主流程，附上部分源碼供大家參考，接下來分析下幾個非常重要流程的具體實(shí)現(xiàn)原理。

客戶端緩存存儲模型

從上圖可以看出，一條消息首先需要確定要被存儲到那個 partition 對應(yīng)的雙端隊(duì)列上；其次，存儲消息的雙端隊(duì)列是以批的維度存儲的，即 N 條消息組成一批，一批消息最多存儲 N 條，超過后則新建一個組來存儲新消息；其次，新來的消息總是從左側(cè)寫入，即越靠左側(cè)的消息產(chǎn)生的時(shí)間越晚；最后，只有當(dāng)一批消息湊夠 N 條后才會發(fā)送給 Broker，否則不會發(fā)送到 Broker 上。

了解了客戶端存儲模型后，來探討下確定消息的 partition（分區(qū)）位置？

確定消息的 partition 位置

消息可分為兩種，一種是指定了 key 的消息，一種是沒有指定 key 的消息。

對于指定了 key 的消息，partition 位置的計(jì)算方式為：Utils.murmur2(key) % numPartitions，即先對 key 進(jìn)行哈希計(jì)算，然后在于 partition 個數(shù)求余，從而得到該條消息應(yīng)該被存儲在哪個 partition 上。

對于沒有指定 key 的消息，partition 位置的計(jì)算方式為：采用 round-robin 方式確定 partition 位置，即采用輪詢的方式，平均的將消息分布到不同的 partition 上，從而避免某些 partition 數(shù)據(jù)量過大影響 Broker 和消費(fèi)端性能。

注意

由于 partition 有主副的區(qū)分，此處參與計(jì)算的 partition 數(shù)量是當(dāng)前有主 partition 的數(shù)量，即如果某個 partition 無主的時(shí)候，則此 partition 是不能夠進(jìn)行數(shù)據(jù)寫入的。

稍微解釋一下，主副 partition 的機(jī)制是為了提高 kafka 系統(tǒng)的容錯性的，即當(dāng)某個 Broker 意外宕機(jī)時(shí)，在此 Broker 上的主 partition 狀態(tài)為不可讀寫時(shí)（只有主 partition 可對外提供讀寫服務(wù)，副 partition 只有數(shù)據(jù)備份的功能），kafka 會從主 partition 對應(yīng)的 N 個副 partition 中挑選一個，并將其狀態(tài)改為主 partition，從而繼續(xù)對外提供讀寫操作。

消息被確定分配到某個 partition 對應(yīng)記錄收集器（即雙端隊(duì)列）后，接下來，發(fā)送線程（Sender）從記錄收集器中收集滿足條件的批數(shù)據(jù)發(fā)送給 Broker，那么發(fā)送線程是如何收集滿足條件的批數(shù)據(jù)的？批數(shù)據(jù)是按照 partition 維度發(fā)送的還是按照 Broker 維度發(fā)送數(shù)據(jù)的？

發(fā)送線程的工作原理

Sender 線程的主要工作是收集滿足條件的批數(shù)據(jù)，何為滿足條件的批數(shù)據(jù)？緩存數(shù)據(jù)是以批維度存儲的，當(dāng)一批數(shù)據(jù)量達(dá)到指定的 N 條時(shí)，就滿足發(fā)送給 Broker 的條件了。

partition 維度和 Broker 維度發(fā)送消息模型對比。

從圖中可以看出，左側(cè)按照 partition 維度發(fā)送消息，每個 partition 都需要和 Broker 建連，總共發(fā)生了四次網(wǎng)絡(luò)連接。而右側(cè)將分布在同一個 Broker 的 partition 按組聚合后在與 Broker 建連，只需要兩次網(wǎng)絡(luò)連接即可。所以 Kafka 選擇右側(cè)的方式。

Sender 的主要工作

第一步：掃描記錄收集器中滿足條件的批數(shù)據(jù)，然后將 partition -> 批數(shù)據(jù)映射轉(zhuǎn)換成 BrokerId -> N 批數(shù)據(jù)的映射。第二步：Sender 線程會為每個 BrokerId 創(chuàng)建一個客戶端請求，然后將請求交給 NetWorkClient，由 NetWrokClient 去真正發(fā)送網(wǎng)絡(luò)請求到 Broker。

NetWorkClient 的工作內(nèi)容

Sender 線程準(zhǔn)備好要發(fā)送的數(shù)據(jù)后，交由 NetWorkClient 來進(jìn)行網(wǎng)絡(luò)相關(guān)操作。主要包括客戶端與服務(wù)端的建連、發(fā)送客戶端請求、接受服務(wù)端響應(yīng)。完成如上一系列的工作主要由如下方法完成。

1. reday()方法。從記錄收集器獲取準(zhǔn)備完畢的節(jié)點(diǎn)，并連接所有準(zhǔn)備好的節(jié)點(diǎn)。
2. send()方法。為每個節(jié)點(diǎn)創(chuàng)建一個客戶端請求，然后將請求暫時(shí)存到節(jié)點(diǎn)對應(yīng)的 Channel（通道）中。
3. poll()方法。該方法會真正輪詢網(wǎng)絡(luò)請求，發(fā)送請求給服務(wù)端節(jié)點(diǎn)和接受服務(wù)端的響應(yīng)。

總結(jié)

以上，即為生產(chǎn)者客戶端的一條消息從生產(chǎn)到發(fā)送到 Broker 上的全過程?，F(xiàn)在是不是就很清晰了呢？也許有些朋友會比較疑惑它的網(wǎng)絡(luò)請求模型是什么樣的，作者就猜你會你會問，下一篇我們就來扒開它的神秘面紗看看其究竟是怎么實(shí)現(xiàn)的，敬請期待。

特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：
長按訂閱更多精彩▼
如有收獲，點(diǎn)個在看，誠摯感謝