概要

繼上一篇講Producer原理的文章過(guò)去已經(jīng)一個(gè)多月了，今天來(lái)講講Consumer的原理。
其實(shí)源碼早就讀了部分了，但是最近工作比較忙，一直沒(méi)空寫(xiě)文章。

整體架構(gòu)流程

技術(shù)名詞解釋

• coordinator：Consumer協(xié)調(diào)器，負(fù)責(zé)管理Consumer需要加入到哪個(gè)消費(fèi)組、消費(fèi)哪個(gè)partition、提交offset等操作
• fetcher：主要作用是獲取待消費(fèi)的records，也是Consumer端最重要的組件
• keyDeserializer：對(duì)record中的key進(jìn)行反序列化
• valueDeserializer：對(duì)record中的value進(jìn)行反序列化
• client：執(zhí)行RPC請(qǐng)求時(shí)的網(wǎng)絡(luò)client，當(dāng)然會(huì)包括一些Kafka內(nèi)部的操作

技術(shù)細(xì)節(jié)

coordinator

其實(shí)協(xié)調(diào)器對(duì)于Consumer的處理分為幾個(gè)階段：

1. Consumer加入的時(shí)候：負(fù)責(zé)判斷Consumer加入到哪個(gè)Consumer group、協(xié)調(diào)消費(fèi)哪個(gè)partition
2. Consumer消費(fèi)過(guò)程中：負(fù)責(zé)記錄Consumer消費(fèi)的partition的元數(shù)據(jù)、partition的消費(fèi)狀態(tài)、消費(fèi)offset；更新partition的offset

fetcher

從Fetcher的數(shù)據(jù)結(jié)構(gòu)里其實(shí)就可以猜到它的作用：緩存已Fetch到的records、去fetch更多的records

• completedFetch：每次fetch請(qǐng)求得到的數(shù)據(jù)，拆分到topicPartition維度。因?yàn)閒etch請(qǐng)求是基于server的node維度，請(qǐng)求回來(lái)的數(shù)據(jù)按照tp維度拆分，得到不同的completedFetch
• completedFetchs: 已經(jīng)fetch到的所有completedFetch
• nextInLineRecords：當(dāng)前正在被消費(fèi)消息的completedFetch對(duì)應(yīng)的所有records，由于對(duì)于同一個(gè)tp，當(dāng)時(shí)Producer發(fā)消息時(shí)，是按照batch維度發(fā)送的，所以此時(shí)completedFetch里也包含多個(gè)batch，每個(gè)batch包含多個(gè)record，也就是records
  如果緩存里沒(méi)有消息呢？
  也就是completedFetchs和nextInLineRecords都是空

client

類(lèi)型是ConsumerNetworkClient，里面包含了一個(gè)NetWorkClient。至于NetWorkClient是如何進(jìn)行數(shù)據(jù)處理及RPC的，可以參考Producer原理解析那篇文章

• unsent：保存的是當(dāng)前需要發(fā)送的fetchRequest
• pendingCompletion：需要被處理的已完成的請(qǐng)求，其實(shí)也就是之前的fetchRequest的response
• client：該client是NetWorkClient，Producer端是直接使用了該client
  所以ConsumerNetworkClient的主要作用：1. 處理之前fetch回來(lái)的數(shù)據(jù)；2. 調(diào)用NetWorkClient將當(dāng)前的fetchRequest發(fā)送出去

consumer#poll的主要流程

1. 判斷是否需要commit offset（默認(rèn)情況下，5秒進(jìn)行一次異步offset的commit）

2. 讀取Fetcher的緩存，如果有數(shù)據(jù)，直接跳轉(zhuǎn)到5

3. 緩存里沒(méi)有數(shù)據(jù)，基于coordinator里保存的partition元數(shù)據(jù)，封裝fetchRequest
   

4. 執(zhí)行client#poll：1. 處理之前fetch回來(lái)的數(shù)據(jù)，解析為completedFetchs；2. 調(diào)用NetWorkClient將當(dāng)前的fetchRequest發(fā)送出去；
   

5. 調(diào)用自定義的消費(fèi)邏輯（程序員自己寫(xiě)的Consumer），處理records

全局總覽

小結(jié)

可以看到Consumer和Producer在邏輯處理上還是有較大不同的。

組件	處理請(qǐng)求	處理方式
producer	主要處理發(fā)送消息。對(duì)應(yīng)RPC，主要是寫(xiě)請(qǐng)求	將業(yè)務(wù)邏輯和IO邏輯解耦。業(yè)務(wù)邏輯：組裝batch；IO邏輯：基于batch組裝request并發(fā)送request
consumer	既要發(fā)送fetchRequest，同時(shí)還要處理fetchResponse。對(duì)于RPC，讀寫(xiě)請(qǐng)求都占比較大	業(yè)務(wù)邏輯和IO邏輯解耦，但是串行化。業(yè)務(wù)邏輯：從fetcher里poll已經(jīng)fetch到的數(shù)據(jù)；IO邏輯：基于partition元數(shù)據(jù)組裝fetchRequest，處理fetchResponse，發(fā)送fetchRequest

Producer的IO是一個(gè)Sender線程在異步運(yùn)行，為什么Consumer不這么干呢？
筆者覺(jué)得原因是：
Producer的邏輯是把消息往外發(fā)，所以Sender運(yùn)行的越快，client這邊為了維護(hù)batch而消耗的資源（內(nèi)存和CPU越少）；而如果Consumer也這么干，實(shí)際消費(fèi)速度趕不上fetch速度的話，會(huì)需要額外的內(nèi)存和CPU資源來(lái)維持更多的completedFetchs，更別說(shuō)如果發(fā)生了rebalance的話，fetch過(guò)來(lái)的completedFetchs可能都是白fetch了。所以，總結(jié)下：1. 兼顧消費(fèi)速度；2. 兼顧client的資源消耗&性能