前言

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

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

1. 客戶端組件

2. 客戶端緩存存儲(chǔ)模型

3. 確定消息的 partition 位置

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

客戶端組件

• KafkaProducer:

KafkaProducer 是一個(gè)生產(chǎn)者客戶端的進(jìn)程，通過(guò)該對(duì)象啟動(dòng)生產(chǎn)者來(lái)發(fā)送消息。

• RecordAccumulator:

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

• Sender:

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

• Selector:

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

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

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

客戶端緩存存儲(chǔ)模型

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

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

確定消息的 partition 位置

消息可分為兩種，一種是指定了 key 的消息，一種是沒(méi)有指定 key 的消息。

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

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

注意

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

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

消息被確定分配到某個(gè) partition 對(duì)應(yīng)記錄收集器（即雙端隊(duì)列）后，接下來(lái)，發(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ù)是以批維度存儲(chǔ)的，當(dāng)一批數(shù)據(jù)量達(dá)到指定的 N 條時(shí)，就滿足發(fā)送給 Broker 的條件了。

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

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

Sender 的主要工作

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

NetWorkClient 的工作內(nèi)容

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

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

總結(jié)

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

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