1.存储结构

我们都知道kafka能堆积非常大的数据，一台服务器，肯定是放不下的。由此出现的集群的概念，集群不仅可以让消息负载均衡，还能提高消息存取的吞吐量。kafka集群中，会有多台broker，每台broker分别在不同的机器上。为了提高吞吐量，每个topic也会都多个分区，同时为了保持可靠性，每个分区还会有多个副本。这些分区副本被均匀的散落在每个broker上，其中每个分区副本中有一个副本为leader，其他的为follower。

在Kafka中主题（Topic）是一个逻辑上的概念，分区（partition）是物理上的存在的。每个partition对应一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端。为防止log文件过大导致数据定位效率低下，Kafka采用了分片和索引机制，将每个partition分为多个segment，每个segment默认1G（ log.segment.bytes ）， 每个segment包括.index文件、.log文件和.timeindex等文件。这些文件位于文件夹下，该文件命名规则为：topic名称+分区号。

• 分区存储：每个Topic分区对应一个目录，目录名为<topic>-<partition>（如test-0）。
• 段文件（Segment）：每个分区的日志被拆分为多个段文件，包括：
• .log文件：存储实际消息。
• .index文件：偏移量索引，映射逻辑偏移量到物理位置。
• .timeindex文件：时间戳索引，映射时间戳到偏移量。
• 文件名规则：以当前段的第一条消息的偏移量命名，用20位数字填充（如00000000000000000000.log）。

Segment的三个文件需要通过特定工具打开才能看到信息

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ../logs/test-3/00000000000000000000.log --print-data-log

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ../logs/test-3/00000000000000000000.index  --print-data-log

• 分段条件：段滚动（Rolling）由以下条件触发：
• 大小阈值：默认1GB（log.segment.bytes）。
• 时间阈值：默认7天（log.roll.hours）。
• 索引文件大小：索引达到一定大小也会触发分段。
• log.segment.bytes: 单个文件段的最大大小, 默认值为1G

• log.roll.ms: 创建新日志段前的最长等待时间（毫秒）。若未设置此参数，则会采用配置项 log.roll.hours 中的数值。 默认值为null

• log.roll.hours: 在创建新日志段之前允许的最大时间（小时），该属性优先级次于 log.roll.ms 属性, 默认值: 168小时也就是7天

2 log文件:

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files ../logs/test-3/00000000000000000000.log --print-data-log

log文件的内容如下:

Dumping ../logs/test-3/00000000000000000000.log
Log starting offset: 0
baseOffset: 0 lastOffset: 9 count: 10 baseSequence: 0 lastSequence: 9 producerId: 1003 producerEpoch: 0 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1742721094962 size: 191 magic: 2 compresscodec: none crc: 3525146444 isvalid: true
| offset: 0 CreateTime: 1742721094923 keySize: -1 valueSize: 6 sequence: 0 headerKeys: [] payload: data-0
| offset: 1 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 1 headerKeys: [] payload: data-1
| offset: 2 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 2 headerKeys: [] payload: data-2
| offset: 3 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 3 headerKeys: [] payload: data-3
| offset: 4 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 4 headerKeys: [] payload: data-4
| offset: 5 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 5 headerKeys: [] payload: data-5
| offset: 6 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 6 headerKeys: [] payload: data-6
| offset: 7 CreateTime: 1742721094961 keySize: -1 valueSize: 6 sequence: 7 headerKeys: [] payload: data-7
| offset: 8 CreateTime: 1742721094962 keySize: -1 valueSize: 6 sequence: 8 headerKeys: [] payload: data-8
| offset: 9 CreateTime: 1742721094962 keySize: -1 valueSize: 6 sequence: 9 headerKeys: [] payload: data-9

通过文件中的magic: 2表示我们kafka消息使用的是V2格式

消息格式演进

• V0格式（Kafka 0.10.0之前）：
• 无时间戳字段，仅支持简单消息结构。
• V1格式（Kafka 0.10.0+）：
• 引入时间戳字段，支持基于时间的日志滚动和索引。
• V2格式（Kafka 0.11.0+）：
• 批处理（Record Batch）：将多条消息打包成一个批次，减少元数据开销。
• 增量CRC：仅对整个批次计算一次CRC，而非每条消息。
• 更紧凑的编码：使用变长整数（Varint）和增量偏移量存储，减少空间占用。

而V2格式的log文件结构:

• Record Batch Header: Kafka消息批次（Record Batch）的元数据摘要
• 后面是每一条记录的详细信息

Record Batch）的元数据属性

3 index文件

索引文件: 记录偏移量（offset）到物理位置（position）的映射

offset: 0 position: 0

index文件中保存的offset为相对offset,这样能确保offset的值所占空间不会过大,因为这样能将offset的值控制在固定大小

当log文件写入4k（这里可以通过log.index.interval.bytes设置）数据，就会写入一条索引信息到index文件中，这样的index索引文件就是一个稀疏索引，它并不会每条日志都建立索引信息。

当Kafka查询一条offset对应实际消息时，可以通过index进行二分查找，获取最近的低位offset，然后从低位offset对应的position开始，从实际的log文件中开始往后查找对应的消息。

Kafka在Log文件中定位offset=899的Record的详细流程:

1. 确定目标Segment文件

1)列出分区目录： 假设Topic分区目录为test-topic-0，包含以下Segment文件：

00000000000000000000.log
00000000000000000000.index
00000000000000000500.log
00000000000000000500.index
00000000000000001000.log
00000000000000001000.index

2)查找包含offset=899的Segment：

• Segment文件名是当前Segment的第一条消息的offset（左闭右开）。
• 目标Segment需满足： Segment起始offset ≤ 899 < 下一个Segment的起始offset。
• 在本例中，00000000000000000000.log的起始offset为0，下一个Segment起始offset为500。 但899 > 500，因此继续查找：
• 下一个Segment起始offset为500，后续为1000。
• 由于500 ≤ 899 < 1000，目标Segment为00000000000000000500.log

2. 使用.index文件快速定位

每个.index文件记录稀疏索引，格式为<offset, physical_position>。 索引条目间隔由index.interval.bytes（默认4KB）控制，并非每条消息都记录。

1)加载.index文件： 读取00000000000000000500.index，假设其内容如下：

offset: 500  position: 0
offset: 600  position: 4096
offset: 800  position: 8192
offset: 900  position: 12288

2)二分查找最近的索引条目：

• 查找小于等于899的最大offset。
• 在示例中，符合条件的是offset=800，对应的position=8192。
• 这一步时间复杂度为O(log N)，N为索引条目数。

3. 从.log文件顺序扫描

根据索引找到的position=8192，从.log文件的该位置开始顺序扫描，直到找到offset=899。

1)定位到物理位置8192： 使用文件指针跳转到00000000000000000500.log的8192字节处。

2)逐条解析消息：

• 读取消息的Offset字段，直到找到offset=899。
• 位置8192: offset=800 → 跳过
  位置8236: offset=801 → 跳过
  ...
  位置9216: offset=899 → 命中目标

4 timeindex文件

时间戳索引文件，它的作用是可以查询某一个时间段内的消息，它的数据结构是：时间戳（8byte）+ 相对offset（4byte），如果要使用这个索引文件，先要通过时间范围找到对应的offset，然后再去找对应的index文件找到position信息，最后在遍历log文件，这个过程也是需要用到index索引文件的。

timestamp: 1742721094962 offset: 9