快捷導(dǎo)航

滴滴二面之Kafka如何讀寫(xiě)副本消息的

更新時(shí)間：2022年01月21日 11:27:17 作者：JavaEdge.

這篇文章主要給大家介紹了關(guān)于滴滴二面之Kafka如何讀寫(xiě)副本消息的相關(guān)資料,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

前言

無(wú)論是讀取副本還是寫(xiě)入副本，都是通過(guò)底層的Partition對(duì)象完成的，而這些分區(qū)對(duì)象全部保存在上節(jié)課所學(xué)的allPartitions字段中?？梢哉f(shuō)，理解這些字段的用途，是后續(xù)我們探索副本管理器類功能的重要前提。

現(xiàn)在，我們就來(lái)學(xué)習(xí)下副本讀寫(xiě)功能。整個(gè)Kafka的同步機(jī)制，本質(zhì)上就是副本讀取+副本寫(xiě)入，搞懂了這兩個(gè)功能，你就知道了Follower副本是如何同步Leader副本數(shù)據(jù)的。

appendRecords-副本寫(xiě)入

向副本底層日志寫(xiě)入消息的邏輯就實(shí)現(xiàn)在ReplicaManager#appendRecords。

Kafka需副本寫(xiě)入的場(chǎng)景：

生產(chǎn)者向Leader副本寫(xiě)入消息
Follower副本拉取消息后寫(xiě)入副本
僅該場(chǎng)景調(diào)用Partition對(duì)象的方法，其余3個(gè)都是調(diào)用appendRecords完成
消費(fèi)者組寫(xiě)入組信息
事務(wù)管理器寫(xiě)入事務(wù)信息（包括事務(wù)標(biāo)記、事務(wù)元數(shù)據(jù)等）

appendRecords方法將給定的一組分區(qū)的消息寫(xiě)入對(duì)應(yīng)Leader副本，并根據(jù)PRODUCE請(qǐng)求中acks的設(shè)置，有選擇地等待其他副本寫(xiě)入完成。然后，調(diào)用指定回調(diào)邏輯。

appendRecords向副本日志寫(xiě)入消息的過(guò)程：

執(zhí)行流程

可見(jiàn)，appendRecords：

實(shí)現(xiàn)消息寫(xiě)入的方法是appendToLocalLog

判斷是否需要等待其他副本寫(xiě)入的方法delayedProduceRequestRequired

appendToLocalLog寫(xiě)入副本本地日志

利用Partition#appendRecordsToLeader寫(xiě)入消息集合，就是利用appendAsLeader方法寫(xiě)入本地日志的。

delayedProduceRequestRequired

判斷消息集合被寫(xiě)入到日志之后，是否需要等待其它副本也寫(xiě)入成功：

private def delayedProduceRequestRequired(
  requiredAcks: Short,
  entriesPerPartition: Map[TopicPartition, MemoryRecords],
  localProduceResults: Map[TopicPartition, LogAppendResult]): Boolean = {
  requiredAcks == -1 && entriesPerPartition.nonEmpty && 
    localProduceResults.values.count(_.exception.isDefined) < entriesPerPartition.size
}

若等待其他副本的寫(xiě)入，須同時(shí)滿足：

requiredAcks==-1
依然有數(shù)據(jù)尚未寫(xiě)完
至少有一個(gè)分區(qū)的消息，已成功被寫(xiě)入本地日志

2和3可結(jié)合來(lái)看。若所有分區(qū)的數(shù)據(jù)寫(xiě)入都不成功，則可能出現(xiàn)嚴(yán)重錯(cuò)誤，此時(shí)應(yīng)不再等待，而是直接返回錯(cuò)誤給發(fā)送方。

而有部分分區(qū)成功寫(xiě)入，部分分區(qū)寫(xiě)入失敗，則可能偶發(fā)的瞬時(shí)錯(cuò)誤導(dǎo)致。此時(shí)，不妨將本次寫(xiě)入請(qǐng)求放入Purgatory，給個(gè)重試機(jī)會(huì)。

副本讀取：fetchMessages

ReplicaManager#fetchMessages負(fù)責(zé)讀取副本數(shù)據(jù)。無(wú)論：

Java消費(fèi)者
APIFollower副本

拉取消息的主途徑都是向Broker發(fā)FETCH請(qǐng)求，Broker端接收到該請(qǐng)求后，調(diào)用fetchMessages從底層的Leader副本取出消息。

fetchMessages也可能會(huì)延時(shí)處理FETCH請(qǐng)求，因Broker端必須要累積足夠多數(shù)據(jù)后，才會(huì)返回Response給請(qǐng)求發(fā)送方。

整個(gè)方法分為：

讀取本地日志

首先判斷，讀取消息的請(qǐng)求方，就能確定可讀取的范圍了。

fetchIsolation，讀取隔離級(jí)別:

對(duì)Follower副本，它能讀取到Leader副本LEO值以下的所有消息
普通Consumer，只能“看到”Leader副本高水位值以下的消息

確定可讀取范圍后，調(diào)用readFromLog讀取本地日志上的消息數(shù)據(jù)，并將結(jié)果賦給logReadResults變量。readFromLog調(diào)用readFromLocalLog，在待讀取分區(qū)上依次調(diào)用其日志對(duì)象的read方法執(zhí)行實(shí)際的消息讀取。

根據(jù)讀取結(jié)果確定Response

根據(jù)上一步讀取結(jié)果創(chuàng)建對(duì)應(yīng)Response：

根據(jù)上一步得到的讀取結(jié)果，統(tǒng)計(jì)可讀取的總字節(jié)數(shù)，然后判斷此時(shí)是否能夠立即返回Reponse。

副本管理器讀寫(xiě)副本的兩個(gè)方法appendRecords和fetchMessages本質(zhì)上在底層分別調(diào)用Log的append和read方法，以實(shí)現(xiàn)本地日志的讀寫(xiě)操作。完成讀寫(xiě)操作后，這兩個(gè)方法還定義了延時(shí)處理的條件。一旦滿足延時(shí)處理?xiàng)l件，就交給對(duì)應(yīng)Purgatory處理。

從這倆方法可見(jiàn)單個(gè)組件融合一起的趨勢(shì)。雖然我們學(xué)習(xí)單個(gè)源碼文件的順序是自上而下，但串聯(lián)Kafka主要組件功能的路徑卻是自下而上。

如副本寫(xiě)入操作，日志對(duì)象append方法被上一層的Partition對(duì)象中的方法調(diào)用，而后者又進(jìn)一步被副本管理器中的方法調(diào)用。我們按自上而下閱讀了副本管理器、日志對(duì)象等單個(gè)組件的代碼，了解了各自的獨(dú)立功能。

現(xiàn)在開(kāi)始慢慢地把它們?nèi)诤弦黄?，?gòu)建Kafka操作分區(qū)副本日志對(duì)象的完整調(diào)用路徑。同時(shí)采用這兩種方式來(lái)閱讀源碼，就能更高效弄懂Kafka原理。

總結(jié)

Kafka副本狀態(tài)機(jī)類ReplicaManager讀寫(xiě)副本的核心方法：

appendRecords：向副本寫(xiě)入消息，利用Log#append方法和Purgatory機(jī)制實(shí)現(xiàn)Follower副本向Leader副本獲取消息后的數(shù)據(jù)同步操作
fetchMessages：從副本讀取消息，為普通Consumer和Follower副本所使用。當(dāng)它們向Broker發(fā)送FETCH請(qǐng)求時(shí)，Broker上的副本管理器調(diào)用該方法從本地日志中獲取指定消息