AbstractFetcherThread：拉取訊息的步驟

副本機制是Kafka實作資料高可靠性的基礎：同一個磁區下的多個副本分散在不同的Broker機器上，它們保存相同的訊息資料以實作高可靠性，那如何確保所有副本上的資料一致性呢？最常見方案當屬Leader/Follower備份機制（Leader/Follower Replication），Kafka磁區的：

• 某個副本會被指定為Leader，負責回應客戶端的讀、寫請求

• 其他副本自動成為Follower，被動同步Leader副本中的資料

  被動同步：Follower副本不斷向Leader副本發送讀取請求，以獲取Leader處寫入的最新訊息資料

本文就研究Follower副本如何通過拉取執行緒實作這一目標，Follower副本在副本同步程序中，還可能發生截斷（Truncation），其原理又是為何？

案例

這部分原始碼貼近底層設計架構原理，閱讀它對我實際有啥用？

生產環境曾發現，一旦Broker上副本數過多，Broker記憶體占用就會很高，HeapDump后，發現在于ReplicaFetcherThread#buildFetch有這么一行代碼：

val builder = fetchSessionHandler.newBuilder()

內部會實體化一個LinkedHashMap，若磁區數很多，該Map會被擴容數次，帶來大量不必要的資料拷貝，既增加記憶體Footprint，又浪費CPU，后續通過將負載轉移到其他Broker解決該問題，

Kafka社區也發現了這個Bug，所以現在變成：

修改后陳述句直接傳入FETCH請求中總的磁區數，并直接將其傳給LinkedHashMap，避免再執行擴容，

說回Follower副本從Leader副本拉取資料，Kafka就是通過ReplicaFetcherThread，副本獲取執行緒實作的訊息拉取及處理，

本文先從抽象基類AbstractFetcherThread研究，最終徹底搞明白Follower端同步Leader端訊息的原理，

AbstractFetcherThread

抽象類，從Broker獲取多個磁區的訊息資料，至于獲取之后如何對這些資料進行處理，則交由子類來實作，

類定義及欄位

除了構造器的這幾個欄位，AbstractFetcherThread還定義了兩個type型別，關鍵字type定義一個型別，可當做一個快捷方式，如FetchData：

type FetchData = FetchResponse.PartitionData[Records]

類似快捷方式：凡原始碼用到FetchResponse.PartitionData[Records]，都可使用FetchData替換，EpochData同理，

FetchData定義里的PartitionData型別，是客戶端clients工程中FetchResponse類的嵌套類，FetchResponse類封裝的是FETCH請求的Response物件，其內PartitionData是個POJO，保存Response中單個磁區資料拉取的各項資料：

• 從該磁區的Leader副本拉取回來的訊息

• 該磁區的高水位值

• 日志起始位移值

• …

在PartitionData中，最需關注的是recordSet，保存了實際的訊息集合，

• 注意到EpochData定義位置，它也是PartitionData型別，但EpochData的PartitionData是OffsetsForLeaderEpochRequest的PartitionData型別

  Kafka原始碼有很多名為PartitionData的嵌套類，很多請求型別中的資料都是按磁區層級分組，因此原始碼很自然地在這些請求類中創建同名嵌套類，所以，注意區分PartitionData嵌套類是定義在哪類請求中的！

磁區讀取狀態類

AbstractFetcherThread構造器中，還有個**PartitionStates[PartitionFetchState]**型別的欄位：

• 泛型引數型別PartitionFetchState類，表征磁區讀取狀態，保存磁區的已讀取位移值和對應副本狀態，

這里的狀態有二：

副本讀取狀態

副本讀取狀態由ReplicaState介面表示：

磁區讀取狀態：

• 可獲取，表明副本獲取執行緒當前能夠讀取資料，
• 截斷中，表明磁區副本正在執行截斷操作（比如該副本剛剛成為Follower副本），
• 被推遲，表明副本獲取執行緒獲取資料時出現錯誤，需要等待一段時間后重試，

磁區讀取狀態中的【可獲取、截斷中】與副本讀取狀態的【獲取中、截斷中】并非嚴格對應，副本讀取狀態處獲取中，并不一定表示磁區讀取狀態就是可獲取狀態，對于磁區，它是否能被獲取的條件要比副本嚴格，

副本獲取執行緒做的事情，日志截斷和訊息獲取：

• isReplicaInSync，副本限流，出鏡率不高

• isDelayed，判斷是否需要推遲獲取對應磁區的訊息

  原始碼會不斷調整那些不需要推遲的磁區的讀取順序，以保證讀取公平性，公平性實作在partitionStates欄位的PartitionStates類，定義在clients工程，會接收一組要讀取的主題磁區，然后輪詢讀取這些磁區以確保公平性，

clients端原始碼自行查閱，

public class PartitionStates<S> {
    private final LinkedHashMap<TopicPartition, S> map = new LinkedHashMap<>();
    ......
    public void updateAndMoveToEnd(TopicPartition topicPartition, S state) {
      map.remove(topicPartition);
      map.put(topicPartition, state);
      updateSize();
    }
    ......
}

PartitionStates輪詢處理要讀取的多個磁區，依靠LinkedHashMap保存所有主題磁區，其元素有明確迭代順序，默認為元素插入的順序，

假設Kafka要讀5個磁區的訊息：A、B、C、D和E，若插入順序：ABCDE，則首先讀磁區A，一旦A被讀取后，為確保各磁區都有同等機會被讀取，代碼需將A插入到磁區串列的最后一位，這就是updateAndMoveToEnd：把A從map中移除，再插回去，這樣A自然就處于串列的最后一位了，這便是PartitionStates的作用，

core API

processPartitionData、truncate、buildFetch和doWork，涵蓋拉取執行緒所做的最重要的3件事：

• 構建FETCH請求

• 執行截斷操作

• 處理拉取后的結果

doWork串聯起前面的這3方法，

最重要的processPartitionData，用于處理讀取回來的訊息集合，它是個抽象方法，因此需子類實作它的邏輯，具體到Follower副本而言， 由ReplicaFetcherThread類實作：

protected def processPartitionData(
  topicPartition: TopicPartition,  // 讀取哪個磁區的資料
  fetchOffset: Long,               // 讀取到的最新位移值
  partitionData: FetchData         // 讀取到的磁區訊息資料
): Option[LogAppendInfo]           // 寫入已讀取訊息資料前的元資料

回傳值Option[LogAppendInfo]：

• 對Follower副本讀訊息寫入日志，可忽略Option，因為肯定會回傳具體LogAppendInfo實體，而不是None
• LogAppendInfo類封裝了很多訊息資料被寫入到日志前的重要元資料資訊，如首條訊息的位移值、最后一條訊息位移值、最大時間戳等

truncate

protected def truncate(
  topicPartition: TopicPartition, // 要對哪個磁區下副本執行截斷操作
  truncationState: OffsetTruncationState  // Offset + 截斷狀態
): Unit

OffsetTruncationState類告訴Kafka要把指定磁區下副本截斷到哪個位移值，封裝了：

• 一個位移值

• 一個截斷完成與否的布林值狀態

buildFetch

protected def buildFetch(
  // 一組要讀取的磁區串列
  // 磁區是否可讀取取決于PartitionFetchState中的狀態
  partitionMap: Map[TopicPartition, PartitionFetchState]): 
// 封裝FetchRequest.Builder物件
ResultWithPartitions[Option[ReplicaFetch]]

本質為指定磁區構建對應FetchRequest.Builder物件，而該物件是構建FetchRequest的核心組件，Kafka中任何型別的訊息讀取，都是通過給指定Broker發送FetchRequest請求來完成的，

doWork

串聯前面3個方法的主要入口方法，

總結

本文研究Kafka的副本同步機制和副本管理器組件，Kafka副本間的訊息同步依賴ReplicaFetcherThread執行緒，AbstractFetcherThread作為拉取執行緒的公共基類，AbstractFetcherThread類定義了很多重要方法，

• AbstractFetcherThread類：拉取執行緒的抽象基類，它定義了公共方法處理所有拉取執行緒的共同邏輯，如執行截斷操作，獲取訊息，

• 拉取執行緒邏輯：回圈執行截斷操作和獲取資料操作，

• 磁區讀取狀態：當前，原始碼定義了3類磁區讀取狀態，拉取執行緒只能拉取處于可讀取狀態的磁區的資料