系列文章目錄

https://zhuanlan.zhihu.com/p/367683572

• 系列文章目錄
• 一. 業務模型
  • 1.1 概念梳理
  • 1.2 檔案分析
    • 1.2.1 資料目錄
    • 1.2.2 .log檔案
    • 1.2.3 .index和.timeindex檔案
  • 1.3 順序IO
  • 1.4 端到端壓縮
• 二. 原始碼結構
  • 2.1 核心類
    • 2.1.1 核心類之間的關系
    • 2.1.1 資料傳遞物件
    • 2.1.2 ReplicaManager
    • 2.1.3 Partition
    • 2.1.4 Replica
    • 2.1.5 Log
    • 2.1.6 LogSegment
    • 2.1.7 OffsetIndex和TimeIndex
  • 2.2 訊息寫入流程
    • 2.2.1 ReplicaManager.appendRecords
    • 2.2.2 Partition.appendRecordsToLeader
    • 2.2.3 Log.appendAsLeader
    • 2.2.4 LogSegment.append
• 三. 總結

一. 業務模型

在上一篇文章中，我們分析了生產者的原理，下一步我們來分析下提交上來的訊息在Server端時如何存盤的，

1.1 概念梳理

Kafka用Topic將資料劃分成內聚性較強的子集，Topic內部又劃分成多個Partition，不過這兩個都是邏輯概念，真正存盤檔案的是Partition所對應的一個或多個Replica，即副本，在存盤層有個概念和副本一一對應——Log，為了防止Log過大，增加訊息過期和資料檢索的成本，Log又會按一定大小劃分成"段"，即LogSegment，用一張圖匯總這些概念間的關系：

1.2 檔案分析

1.2.1 資料目錄

Kafkap組態檔(server.properties)中有一個配置項——log.dir，其指定了kafka資料檔案存放位置，為了研究資料目錄的結構，我們先創建一個Topic(lao-zhang-tou-topic)

kafka-console-producer.sh --topic lao-zhang-tou-topic --bootstrap-server localhost:9092

然后向其中寫幾條訊息

kafka-console-producer.sh --topic lao-zhang-tou-topic --bootstrap-server localhost:9092
{"message":"This is the first message"}
{"message":"This is the sencond message"}

接下來我們來看看log.dir指定目錄下存放了那些檔案

該目錄下檔案分3類：

1. 資料檔案夾

   如截圖中的lao-zhang-tou-topic-0

2. checkpoint檔案

   • cleaner-offset-checkpoint
   • log-start-offset-checkpoint
   • recovery-point-offset-checkpoint
   • replication-offset-checkpoint

3. 組態檔

   meta.properties

第2、3類檔案后續文章會詳細分析，本文主要關注截圖中lao-zhang-tou-topic-0目錄，

實際上，該目錄對應上文提到的Log概念，命名規則為 ${Topic}-${PartitionIndex}，該目錄下，名稱相同的.log檔案、.index檔案、.timeindex檔案構成了一個LogSegment，例如圖中的 00000000000000000000.log、00000000000000000000.index、00000000000000000000.timeindex 三個檔案，其中.log是資料檔案，用于存盤訊息資料；.index和.timeindex是在.log基礎上建立起來的索引檔案，

1.2.2 .log檔案

log檔案將訊息資料依次排開進行存盤

每個Message內部分為"資料頭"(LOG_OVERHEAD)和"資料體"(Record)兩部分

其中，LOG_OVERHEAD包含兩個欄位：

1. offset：每條資料的邏輯偏移量，按插入順序分別為0、1、2... ... N；每個訊息的offset在Partition內部是唯一的；
2. size：資料體(RECORD)部分的長度；

RECORD內部格式如下：

其中，

• crc32：校驗碼，用于驗證資料完整性；

• magic：訊息格式的版本號；v0=0，v1=1；本文講v1格式；

• timestamp：時間戳，具體業務含義依attributes的值而定；

• attributes：屬性值；其 8bits 的含義如下

  

• keyLength：key值的長度；

• key：訊息資料對應的key；

• valueLength：value值的長度；

• value：訊息體，承載業務資訊；

1.2.3 .index和.timeindex檔案

.index檔案是依offset建立其的稀疏索引，可減少通過offset查找訊息時的遍歷資料量，.index檔案的每個索引條目占8 bytes，有兩個欄位：relativeOffset 和 position(各占4 bytes)，也就是訊息offset到其在檔案中偏移量的一個映射，那有人要問了，索引項中保存的明明是一個叫relativeOffset的東西，為什么說是offset到偏移量的映射呢？其實，準確的來講，relativeOffset指的的相對偏移量，是對LogSegment基準offset而言的，我們注意到，一個LogSegment內的.log檔案、.index檔案、和.index檔案除后綴外的名稱都是相同的，其實這個名稱就是該LogSegment的基準offset，即LogSegment內保存的第一條訊息對應的offset，baseOffset + relativeOffset即可得到offset，所以稱索引項是offset到物理偏移量的映射，

不是所有的訊息都對應.index檔案內的一個條目，Kafka會每隔一定量的訊息才會在.index建立索引條目，間隔大小由"log.index.interval.bytes"配置指定，.index檔案布局示意圖如下：

.timeindex檔案和.index原理相同，只不過其IndexEntry的兩個欄位分別為timestamp(8 bytes)和relativeOffset(4 bytes)，用于減少以時間戳查找訊息時遍歷元素數量，

1.3 順序IO

對于我們常用的機械硬碟，其讀取資料分3步：

1. 尋道；
2. 尋找扇區；
3. 讀取資料；

前兩個，即尋找資料位置的程序為機械運動，我們常說硬碟比記憶體慢，主要原因是這兩個程序在拖后腿，不過，硬碟比記憶體慢是絕對的嗎？其實不然，如果我們能通過順序讀寫減少尋找資料位置時讀寫磁頭的移動距離，硬碟的速度還是相當可觀的，一般來講，IO速度層面，記憶體順序IO > 磁盤順序IO > 記憶體隨機IO > 磁盤隨機IO，

Kafka在順序IO上的設計分兩方面看：

1. LogSegment創建時，一口氣申請LogSegment最大size的磁盤空間，這樣一個檔案內部盡可能分布在一個連續的磁盤空間內；
2. .log檔案也好，.index和.timeindex也罷，在設計上都是只追加寫入，不做更新操作，這樣避免了隨機IO的場景；

Kafka是公認的高性能訊息中間件，順序IO在這里占了很大一部分因素，

不知道大家有沒有聽過這樣一個說法：Kafka集群能承載的Partition數量有上限，很大一部分原因是Partition數量太多會抹殺掉Kafka順序IO設計帶來的優勢，相當于自廢武功，Why？因為不同Partition在磁盤上的存盤位置可不保證連續，當以不同Partition為讀寫目標并發地向Kafka發送請求時，Server端近似于隨機IO，

1.4 端到端壓縮

一條壓縮訊息從生產者處發出后，其在消費者處才會被解壓，Kafka Server端不會嘗試決議訊息體，直接原樣存盤，省掉了Server段壓縮&解壓縮的成本，這也是Kafka性能喜人的原因之一，

二. 原始碼結構

2.1 核心類

2.1.1 核心類之間的關系

Kafka訊息存盤涉及的核心類有：

• ReplicaManager
• Partition
• Replica
• Log
• LogSegment
• OffsetIndex
• TimeIndex
• MemoryRecords
• FileRecords

它們之間的關系如下圖：

2.1.1 資料傳遞物件

Kafka訊息存盤的基本單位不是"一條訊息"，而是"一批訊息"，在生產者文章中提到過，Producer針對每個Partition會攢一批訊息，經過壓縮后發到Server端，Server端會將對應Partition下的這一"批"訊息作為一個整體進行管理，所以在Server端，一個"Record"表示"一批訊息"，而資料傳遞物件"XXXRecords"則可以表示一批或多批訊息，

MemoryRecords所表示的訊息資料存盤于記憶體，比如Server端從接到生產者訊息到將訊息存入磁盤的程序就用MemoryRecords來傳遞資料，因為這期間訊息需要暫存于記憶體，且沒有磁盤資料與之對應，MemoryRecords核心屬性有兩個：

FileRecords所表示的訊息資料存盤于磁盤檔案，比如從磁盤讀出訊息回傳給消費者的程序就用FileRecords來傳遞資料，其核心屬性如下：

2.1.2 ReplicaManager

ReplicaManager負責管理本節點存盤的所有副本，這個類的屬性真的巨多，不過不要慌，對于訊息存盤原理這塊，我們只需要關注下面這一個屬性就可以，其他和請求處理以及副本復制相關的屬性我們放到后邊對應章節慢慢分析，

2.1.3 Partition

Partition物件負責維護本磁區下的所有副本，其核心屬性如下：

2.1.4 Replica

Replica負責維護Log物件，Replica是業務模型層面"副本"的表示，Log是資料存盤層面的"副本"，Replica核心屬性如下：

2.1.5 Log

Log負責維護副本下的LogSegment，其核心屬性如下：

2.1.6 LogSegment

LogSegment則實實在在維護訊息資料，其核心屬性如下：

2.1.7 OffsetIndex和TimeIndex

首先兩個索引都繼承于AbstractIndex，那么他們就有一批共同的核心屬性：

實際上，這些屬性已足夠表達當前的索引邏輯，OffsetIndex和TimeIndex均未再額外自定義屬性，

2.2 訊息寫入流程

訊息寫入流程時序圖如下：

需要提一點，這里不是為了讓諸君將這一串流程視為整體記入腦海，面向物件的代碼仍然要從面向物件的角度去理解，所以這里重要的是各個類各自內部的邏輯，這有助于進一步明確類所扮演的角色，

2.2.1 ReplicaManager.appendRecords

def appendRecords(timeout: Long,
                    requiredAcks: Short,
                    internalTopicsAllowed: Boolean,
                    isFromClient: Boolean,
                    entriesPerPartition: Map[TopicPartition, MemoryRecords],// 各Partition上待插入的訊息資料
                    responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
                    delayedProduceLock: Option[Lock] = None,
                    recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => ()) {
      ... ...
      val localProduceResults = appendToLocalLog(internalTopicsAllowed = internalTopicsAllowed,
        isFromClient = isFromClient, entriesPerPartition, requiredAcks)
      ... ...
    
}

private def appendToLocalLog(internalTopicsAllowed: Boolean,
                               isFromClient: Boolean,
                               entriesPerPartition: Map[TopicPartition, MemoryRecords],
                               requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
   	  ... ...
      // step1 reject appending to internal topics if it is not allowed
      if (Topic.isInternal(topicPartition.topic) && !internalTopicsAllowed) {
        (topicPartition, LogAppendResult(
          LogAppendInfo.UnknownLogAppendInfo,
          Some(new InvalidTopicException(s"Cannot append to internal topic ${topicPartition.topic}"))))
      } else {
        try {
          //step2 若本Broker節點不承載對應partition的主副本, 這步會拋例外
          val (partition, _) = getPartitionAndLeaderReplicaIfLocal(topicPartition)
          //step3 將訊息寫入對應Partition主副本, 并喚醒相關的等待操作(比如, 消費等待)
          val info = partition.appendRecordsToLeader(records, isFromClient, requiredAcks)
          ... ...
        }
      }
    }
}

appendRecords直接呼叫appendToLocalLog，后者才是真正實行邏輯的方法，ReplicaManager的邏輯基本分三步走：

1. 檢查目標Topic是否為Kafka內部Topic，若是的話根據配置決定是否允許寫入；
2. 獲取對應的Partition物件；
3. 呼叫Partition.appendRecordsToLeader寫入訊息資料；

2.2.2 Partition.appendRecordsToLeader

接下來看看Partition內部的邏輯

def appendRecordsToLeader(records: MemoryRecords, isFromClient: Boolean, requiredAcks: Int = 0): LogAppendInfo = {
    val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock) {
      leaderReplicaIfLocal match {
        //step1 判斷Leader副本是否在當前節點
        case Some(leaderReplica) =>
          //step2 獲取Log物件
          val log = leaderReplica.log.get
          ... ...
          //step3 呼叫Log物件方法寫入資料
          val info = log.appendAsLeader(records, leaderEpoch = this.leaderEpoch, isFromClient)
          ... ...

        // 若本節點不是目標Partition的Leader副本, 拋例外
        case None =>
          throw new NotLeaderForPartitionException("Leader not local for partition %s on broker %d"
            .format(topicPartition, localBrokerId))
      }
    }
    ... ...
  }

這里的邏輯也分3步走：

1. 判斷Leader副本是否在當前節點；
2. 獲取Log物件；
3. 呼叫Log物件的appendAsLeader方法寫入資料；

這里我們額外看下第1步的原理，leaderReplicaIfLocal是個方法

def leaderReplicaIfLocal: Option[Replica] =
  leaderReplicaIdOpt.filter(_ == localBrokerId).flatMap(getReplica)

def getReplica(replicaId: Int = localBrokerId): Option[Replica] = Option(allReplicasMap.get(replicaId))

其核心思想是那本節點BrokerId和Leader副本所在節點的BrokerId作比較，若相等，則回傳對應的Replica物件，

2.2.3 Log.appendAsLeader

def appendAsLeader(records: MemoryRecords, leaderEpoch: Int, isFromClient: Boolean = true): LogAppendInfo = {
  append(records, isFromClient, assignOffsets = true, leaderEpoch)
}

private def append(records: MemoryRecords, isFromClient: Boolean, assignOffsets: Boolean, leaderEpoch: Int): LogAppendInfo = {
  ... ...
  // maybe roll the log if this segment is full
  val segment = maybeRoll(validRecords.sizeInBytes, appendInfo)
  ... ...
  // 將訊息插入segment
  segment.append(largestOffset = appendInfo.lastOffset,
    largestTimestamp = appendInfo.maxTimestamp,
    shallowOffsetOfMaxTimestamp = appendInfo.offsetOfMaxTimestamp,
    records = validRecords)
  ... ...
}

appendAsLeader方法直接呼叫append方法，后者兩步走：

1. 判斷是否需要創建一個新的LogSegment，并回傳最新的LogSegment；
2. 呼叫LogSegment.append方法寫入資料；

這里我們再額外關注下第1步的判斷標準，主要還是根據LogSegment.shouldRoll方法的回傳值來作決策：

def shouldRoll(messagesSize: Int, maxTimestampInMessages: Long, maxOffsetInMessages: Long, now: Long): Boolean = {
  val reachedRollMs = timeWaitedForRoll(now, maxTimestampInMessages) > maxSegmentMs - rollJitterMs

  size > maxSegmentBytes - messagesSize ||
    (size > 0 && reachedRollMs) ||
    offsetIndex.isFull || timeIndex.isFull || !canConvertToRelativeOffset(maxOffsetInMessages)
}

Kafka的原始碼很清晰的，這方面值得點贊和學習，從shouldRoll的結果運算式我們可以看到，以下4類場景中，LogSegment需要向前滾動：

1. 若接受新訊息的寫入，當前LogSegment將超過最大位元組數限制；
2. 若接受新訊息的寫入，當前LogSegment將超過最大時間跨度限制；
3. 當前LogSegment對應的索引已無法寫入新資料；
4. 輸入的offset不在當前LogSegment表示范圍；

2.2.4 LogSegment.append

def append(largestOffset: Long,
             largestTimestamp: Long,
             shallowOffsetOfMaxTimestamp: Long,
             records: MemoryRecords): Unit = {
    // step1.1 判斷輸入訊息大小
    if (records.sizeInBytes > 0) {
      trace(s"Inserting ${records.sizeInBytes} bytes at end offset $largestOffset at position ${log.sizeInBytes} " +
            s"with largest timestamp $largestTimestamp at shallow offset $shallowOffsetOfMaxTimestamp")
      // step1.2 校驗offset
      val physicalPosition = log.sizeInBytes()
      if (physicalPosition == 0)
        rollingBasedTimestamp = Some(largestTimestamp)

      ensureOffsetInRange(largestOffset)

      // step2 append the messages
      val appendedBytes = log.append(records)
      trace(s"Appended $appendedBytes to ${log.file} at end offset $largestOffset")
      // step3 Update the in memory max timestamp and corresponding offset.
      if (largestTimestamp > maxTimestampSoFar) {
        maxTimestampSoFar = largestTimestamp
        offsetOfMaxTimestamp = shallowOffsetOfMaxTimestamp
      }
      // step4 append an entry to the index (if needed)
      if (bytesSinceLastIndexEntry > indexIntervalBytes) {
        offsetIndex.append(largestOffset, physicalPosition)
        timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestamp)
        bytesSinceLastIndexEntry = 0
      }
      bytesSinceLastIndexEntry += records.sizeInBytes
    }
  }

LogSegment.append大體可以分為4步：

1. 資料校驗
   1. 校驗輸入訊息大小；
   2. 校驗offset；
2. 寫入資料(注意: 此步的log物件不是Log類的實體，而是FileRecords的實體)；
3. 更新統計資料；
4. 處理索引；

三. 總結

本文從業務模型&原始碼角度分析了Kafka訊息存盤原理，才疏學淺，不一定很全面，

另外也可以在目錄中找到同系列的其他文章：Kafka原始碼分析系列-目錄(收藏關注不迷路)，

歡迎諸君隨時來交流，

標籤：其他

上一篇：JUC下工具類CountDownLatch用法以及原始碼理解

下一篇：Java模擬面試總結