【Kafka源码解读和使用指南】第43篇：Kafka日志存储源码解析（二）——Segment分段存储的精妙设计

上一篇【第42篇】Kafka日志存储源码解析（一）——消息是怎么被"写入磁盘"的 下一篇：【第44篇】Kafka日志存储源码解析（三）——OffsetIndex稀疏索引的"秘密武器"

摘要

上篇我们了解了消息是如何被顺序写入磁盘的。本文将深入LogSegment——Kafka日志存储的第二层核心抽象。每个Partition在磁盘上被切分成多个Segment，每个Segment包含一对文件：.log（日志文件）和.index（偏移量索引文件）。这种分段设计让Kafka能够在保持高性能的同时，实现高效的日志清理、数据查找和崩溃恢复。本文将详解LogSegment的源码实现、Segment的滚动机制，以及为什么Kafka选择这样的分段策略。

一、为什么需要Segment分段？

在深入源码之前，先理解为什么Kafka要把一个Partition的日志切分成多个Segment？

1.1 单文件的困境

【如果只用单个日志文件】

partition-0/
└── 00000000000000000000.log (500GB!)

问题：
① 文件太大，无法全部mmap到内存
② 清理旧数据时只能遍历整个文件，效率极低
③ 崩溃恢复需要扫描整个文件
④ 索引文件也会无限增大

1.2 Segment分段的解法

【Kafka的分段方案】

partition-0/
├── 00000000000000000000.log (1GB)
├── 00000000000000000000.index
├── 00000000000001048576.log (1GB)
├── 00000000000001048576.index
├── 00000000000002097152.log (1GB)
├── 00000000000002097152.index
└── …

优势：
✓ 每个Segment大小可控，可以mmap到内存
✓ 清理旧数据只需删除整个Segment文件
✓ 崩溃恢复只需扫描最新的几个Segment
✓ 索引文件大小也可控

二、LogSegment核心源码解析

2.1 类结构一览

// LogSegment.scala (核心字段，简化版)
class LogSegment(val log: FileMessageSet, // 对应的日志文件
val index: OffsetIndex, // 对应的偏移量索引
val timeIndex: TimeIndex, // 时间戳索引（0.10+）
val baseOffset: Long, // 此Segment的起始offset
val indexIntervalBytes: Int, // 两个索引项之间的最小字节间隔
val bytesSinceLastIndexEntry: Int = 0) { // 自上次建索引以来的字节数

// 核心方法
def append(firstOffset: Long, messages: ByteBufferMessageSet): Unit
def read(startOffset: Long, maxOffset: Option[Long], maxSize: Int,
maxPosition: Long = Int.MaxValue): FetchDataInfo
def recover(maxMessageSize: Int): Int
def truncateTo(offset: Long): Int
def readNextOffset: Long
}

2.2 append()——向Segment追加消息

这是消息写入的核心路径，每收到一批消息就会调用此方法：

// LogSegment.append() 源码解析
def append(firstOffset: Long, messages: ByteBufferMessageSet): Unit = {
// ① 检查：当前Segment是否已有数据，且第一条消息的offset必须连续
if (log.sizeInBytes > 0) {
// 校验offset的连续性，防止数据损坏
require(lastOffset >= 0 && firstOffset == lastOffset + 1,
"Attempt to append out of order messages")
}

// ② 判断是否需要在写入消息前先建立索引
// 规则：如果距上次建索引已写入超过 indexIntervalBytes 字节，则建索引
if (bytesSinceLastIndexEntry > indexIntervalBytes) {
index.append(firstOffset, log.sizeInBytes)
timeIndex.maybeAppend(messages.maxTimestamp, firstOffset)
bytesSinceLastIndexEntry = 0 // 重置计数器
}

// ③ 将消息追加到.log文件（调用FileMessageSet.append）
log.append(messages)

// ④ 更新字节计数器
bytesSinceLastIndexEntry += messages.sizeInBytes

// ⑤ 更新最大时间戳（用于时间戳索引）
if (messages.maxTimestamp > maxTimestamp)
maxTimestamp = messages.maxTimestamp
}

【append() 执行流程图】

调用 append(firstOffset=100, messages)
│
▼
┌──────────────────────────────────────────────┐
│ ① 校验offset连续性 │
│ lastOffset=99, firstOffset=100 ✓ │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────────┐
│ ② 判断是否需要建索引 │
│ bytesSinceLastIndexEntry=5000 │
│ indexIntervalBytes=4096 │
│ 5000 > 4096 → 需要建索引 ✓ │
│ → index.append(100, currentPosition) │
│ → bytesSinceLastIndexEntry = 0 │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────────┐
│ ③ 追加到.log文件 │
│ log.append(messages) │
│ → FileChannel.write() │
└──────────────┬───────────────────────────┘
│
▼
┌──────────────────────────────────────────────┐
│ ④ 更新计数器 │
│ bytesSinceLastIndexEntry += messages.size │
└──────────────────────────────────────────────┘

2.3 read()——从Segment读取消息

消费者拉取消息时，最终会调用此方法：

// LogSegment.read() 源码解析
def read(startOffset: Long, maxOffset: Option[Long],
maxSize: Int, maxPosition: Long): FetchDataInfo = {

// ① 通过OffsetIndex查找startOffset对应的物理位置
val startPosition = index.lookup(startOffset)
// startPosition = OffsetPosition(physicalOffset, offset)

// ② 从物理位置开始读取消息
val fetchInfo = log.read(startPosition.position, maxSize)

// ③ 封装返回结果
FetchDataInfo(offsetMetadata, fetchInfo)
}

三、Segment滚动机制——何时切新文件？

Segment滚动（Rolling）是指：当前活跃的Segment写满了，需要创建一个新的Segment来接收后续消息。

3.1 滚动触发条件

Kafka在每次append之后都会检查是否需要滚动，触发条件有三个：

【Segment滚动触发条件】

条件1: 当前Segment大小 ≥ log.segment.bytes (默认1GB)
│
▼
条件2: 当前Segment创建时间 ≥ log.roll.ms (默认7天)
│
▼
条件3: 索引文件已满（无法再写入新的索引项）
│
▼
满足以上任一条件 → 触发滚动，创建新的activeSegment

3.2 滚动过程源码

// Log.scala – maybeRoll() 方法（简化版）
def maybeRoll(messagesSize: Int): LogSegment = {
val segment = activeSegment // 当前活跃的Segment

// 检查三个滚动条件
if (segment.size > config.segmentSize – messagesSize ||
segment.timeWaited > config.segmentMs ||
segment.index.isFull) {
// 执行滚动
roll()
} else {
segment // 不需要滚动，返回当前Segment
}
}

def roll(): LogSegment = {
// ① 获取新的baseOffset（即当前LEO）
val newOffset = logEndOffset

// ② 创建新的.log和.index文件
val newSegment = new LogSegment(
FileMessageSet(newOffset, ...),
OffsetIndex(newOffset, ...),
...
)

// ③ 将新Segment加入segments映射表
segments.put(newOffset, newSegment)

// ④ 返回新的activeSegment
newSegment
}

【Segment滚动时序图】

Producer发送消息
│
▼
Log.append()
│
▼
Log.maybeRoll()
│
├── 检查条件1: size > 1GB?
│
├── 检查条件2: age > 7天?
│
└── 检查条件3: index isFull?
│
▼ (任一条件满足)
┌──────────────────────────────────┐
│ 创建新Segment │
│ – 新文件: 00000…N.log │
│ – 新索引: 00000…N.index │
│ – baseOffset = 当前LEO │
└──────────────┬───────────────────┘
│
▼
后续消息写入新的activeSegment

四、recover()——崩溃恢复时重建索引

当Broker崩溃重启后，Kafka需要重建内存中的索引数据（因为索引文件可能不是最新的）。这就是recover()方法的作用。

4.1 为什么需要recover？

【崩溃恢复场景】

正常情况：
写入消息 → 更新.log文件
→ 更新.index文件（内存mmap）
→ 定期msync()刷到磁盘

崩溃时：
.log文件已写入（Page Cache可能没有刷盘，但大部分还在）
.index文件可能没有完全刷盘！
→ 需要重新扫描.log，重建.index

4.2 recover()源码解析

// LogSegment.recover() 源码解析
def recover(maxMessageSize: Int): Int = {
var validBytes = 0
val iter = log.iterator(maxMessageSize) // 遍历.log文件中的所有消息

try {
while (iter.hasNext) {
val entry = iter.next()
val message = entry.message

// 校验消息完整性（CRC32、size等）
message.ensureValid()

// 判断是否需要为此消息建立索引
if (validBytes – lastIndexEntry > indexIntervalBytes) {
// 建索引：记录此offset对应的物理位置
index.append(entry.offset, validBytes)
timeIndex.maybeAppend(message.timestamp, entry.offset)
lastIndexEntry = validBytes
}

validBytes += message.sizeInBytes // 累计有效字节数
}
} catch {
case e: InvalidMessageException =>
// 遇到损坏的消息，停止扫描（后续消息也不可信）
}

// 截断.log文件到validBytes位置（丢弃损坏部分）
log.truncateTo(validBytes)
index.trimToValidSize()

validBytes // 返回有效字节数
}

五、分段设计的精妙之处

5.1 稀疏索引 + 分段 = 高效查找

【查找offset=1030的消息】

步骤1: 在segments中二分查找
→ 找到baseOffset=0的Segment（因为0 ≤ 1030 < 1024? 不对…）
→ 找到baseOffset=1024的Segment

步骤2: 在OffsetIndex中二分查找
→ 找到offset ≤ 1030的最大索引项
→ 假设找到: offset=1028, position=56000

步骤3: 从position=56000开始顺序扫描
→ 读取每条消息的offset，直到找到offset=1030

5.2 分段 vs 不分段的对比

六、核心参数总结

本篇小结

本文深入解析了Kafka LogSegment的源码实现：

• 分段设计：将Partition切分成多个Segment，每个Segment包含.log + .index两个文件
• append()：追加消息时，根据indexIntervalBytes间隔建立稀疏索引
• read()：读取消息时，先通过OffsetIndex查找物理位置，再顺序扫描
• 滚动机制：满足大小/时间/索引满三个条件之一时，创建新的activeSegment
• recover()：崩溃恢复时重建索引，保证索引与日志的一致性

下一篇文章，我们将深入OffsetIndex的源码实现，看看Kafka的稀疏索引是如何在内存中使用mmap加速查找的。

上一篇【第42篇】Kafka日志存储源码解析（一）——消息是怎么被"写入磁盘"的 下一篇：【第44篇】Kafka日志存储源码解析（三）——OffsetIndex稀疏索引的"秘密武器"