Kafka 的消息存储结构：索引文件与数据文件

旧城等待， 2022-09-10 00:07 219阅读 0赞

总在说Kafka，Topic，Key，Value；Kafka client发送消息/消费消息； Kafka Server 会接收生产者的消息，那么 Kafka 接收到消息并刷到磁盘之后。消息文件是如何存储的呢？

## Kafka 的消息存储结构 ##

Kafka 有 Topic 和 Partition 两个概念，一个 Topic 可以有多个 Partition。在实际存储的时候，Topic + Partition 对应一个文件夹，这个文件夹对应的是这个 Partition 的数据。

举例，一个topic为kafka-topic-loongshawn，对应3个分区：

./kafka-topics.sh 
    --describe --zookeeper 127.0.0.1:2181 --topic kafka-loongshawn-01
    Topic:kafka-loongshawn-01    PartitionCount:3    ReplicationFactor:3 Configs:
        Topic: kafka-loongshawn-01   Partition: 0    Leader: 1   Replicas: 1,2,0 Isr: 1,2,0
        Topic: kafka-loongshawn-01   Partition: 1    Leader: 2   Replicas: 2,0,1 Isr: 2,1,0
        Topic: kafka-loongshawn-01   Partition: 2    Leader: 0   Replicas: 0,1,2 Isr: 1,2,0

那么，其数据文件的目录结构如下所示，3个分区对应3个文件夹，文件夹命名topic-分区序号：

# 分区目录文件
    drwxr-x--- 2 root root 4096 Jul 26 19:35 kafka-loongshawn-01-0
    drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-loongshawn-01-1
    drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-loongshawn-01-2

在 Kafka 的数据文件目录下，一个 Partition 对应一个唯一的文件夹。如果有 4 个 Topic，每个 Topic 有 5 个 Partition，那么一共会有 4 \* 5 = 20 个文件夹。以此类推。

而在文件夹下，Kafka 消息是采用 Segment File 的存储方式进行存储的。即每个分区文件夹下拆分为N个小文件，共同存储改分区中的所有消息。

Segment File 的大概意思是：将大文件拆分成小文件来存储，这样一个大文件就变成了一段一段（Segment 段）。这样的好处是 IO 加载速度快，不会有很长的 IO 加载时间。Kafka 的消息存储就采用了这种方式。

# 分区目录中的日志数据文件和日志索引文件
    -rw-r----- 1 root root 512K Jul 24 19:51 00000000000000000000.index
    -rw-r----- 1 root root 1.0G Jul 24 19:51 00000000000000000000.log
    -rw-r----- 1 root root 768K Jul 24 19:51 00000000000000000000.timeindex
    -rw-r----- 1 root root 512K Jul 24 20:03 00000000000022372103.index
    -rw-r----- 1 root root 1.0G Jul 24 20:03 00000000000022372103.log
    -rw-r----- 1 root root 768K Jul 24 20:03 00000000000022372103.timeindex
    -rw-r----- 1 root root 512K Jul 24 20:15 00000000000044744987.index
    -rw-r----- 1 root root 1.0G Jul 24 20:15 00000000000044744987.log
    -rw-r----- 1 root root 767K Jul 24 20:15 00000000000044744987.timeindex
    -rw-r----- 1 root root  10M Jul 24 20:21 00000000000067117761.index
    -rw-r----- 1 root root 511M Jul 24 20:21 00000000000067117761.log
    -rw-r----- 1 root root  10M Jul 24 20:21 00000000000067117761.timeindex

如上所示，在一个文件夹下的数据会根据 Kafka 的配置拆分成多个小文件。拆分规则可以根据文件大小拆分，也可以根据消息条数拆分，这个是 Kafka 的一个配置，这里不细说。

在 Kafka 的数据文件夹下，分为三种类型的文件：索引文件（Index File）、数据文件（Data File）和时间索引（Time File）。索引文件存的是消息的索引信息，帮助快速定位到某条消息。数据文件存储的是具体的消息内容。

### 索引文件 ###

索引文件的命名统一为数字格式，其名称表示 Kafka 消息的偏移量。我们假设索引文件的数字为 N，那么就代表该索引文件存储的第一条 Kafka 消息的偏移量为 N + 1，而上个文件存储的最后一条 Kafka 消息的偏移量为 N（因为 Kafka 是顺序存储的）。

#### 索引文件表象 ####

例如00000000000022372103.index 索引文件，其表示文件存储的第一条 Kafka 消息的偏移量为 22372104。而 22372103表示的是 00000000000000000000.index 这个索引文件的最后一条消息。所以 00000000000022372103.index 索引文件，其存储的 Kafka 消息偏移量范围为 22372104~44744987。

# 分区目录中的日志索引文件
    -rw-r----- 1 root root 512K Jul 24 19:51 00000000000000000000.index
    -rw-r----- 1 root root 512K Jul 24 20:03 00000000000022372103.index
    -rw-r----- 1 root root 512K Jul 24 20:15 00000000000044744987.index
    -rw-r----- 1 root root  10M Jul 24 20:21 00000000000067117761.index

#### 索引文件内容 ####

索引文件存储的是简单地索引数据，其格式为：「N,Position」。其中 N 表示索引文件里的第几条消息，而 Position 则表示该条消息在数据文件（Log File）中的物理偏移地址。例如下图中的「1,365」表示：索引文件里的第 1 条消息（即 offset 22372104 的消息，22372104 = 22372103+1），其在数据文件中的物理偏移地址为 365。

Dumping /apps/Kafka/kafkalogs/kafka-loongshawn-01-0/00000000000022372103.index
    offset: 1 position: 365
    offset: 3 position: 730
    offset: 5 position: 1460
    offset: 7 position: 2825
    offset: 9 position: 4190
    offset: 11 position: 5555
    offset: 13 position: 6584
    offset: 15 position: 7211
    ...

其他的以此类推，例如：「8,3686」表示 offset 为 22372112 的 Kafka 消息，其在数据文件中的物理偏移地址为 3686。

### 数据文件 ###

数据文件的命名格式与索引文件的命名格式完全一样，这里就不再赘述了。

通过上面索引文件的分析，我们已经可以根据 offset 快速定位到某个数据文件了。那接着我们怎么读取到这条消息的内容呢？要读取到这条消息的内容，我们需要搞清楚数据文件的存储格式。

数据文件就是所有消息的一个列表，而每条消息都有一个固定的格式。

message物理结构：  
![在这里插入图片描述][95bfc2f6a9b3481c952ae64233d9a620.png_pic_center]  
参数释义：

<table> 
 <thead> 
  <tr> 
   <th>关键字</th> 
   <th align="left">解释说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>8 byte offset</td> 
   <td align="left">在parition(分区)内的每条消息都有一个有序的id号，这个id号被称为偏移(offset),它可以唯一确定每条消息在parition(分区)内的位置。即offset表示partiion的第多少message</td> 
  </tr> 
  <tr> 
   <td>4 byte message size</td> 
   <td align="left">message大小</td> 
  </tr> 
  <tr> 
   <td>4 byte CRC32</td> 
   <td align="left">用crc32校验message</td> 
  </tr> 
  <tr> 
   <td>1 byte “magic”</td> 
   <td align="left">表示本次发布Kafka服务程序协议版本号</td> 
  </tr> 
  <tr> 
   <td>1 byte “attributes”</td> 
   <td align="left">表示为独立版本、或标识压缩类型、或编码类型。</td> 
  </tr> 
  <tr> 
   <td>4 byte key length</td> 
   <td align="left">表示key的长度,当key为-1时，K byte key字段不填</td> 
  </tr> 
  <tr> 
   <td>K byte key</td> 
   <td align="left">可选</td> 
  </tr> 
  <tr> 
   <td>value bytes payload</td> 
   <td align="left">表示实际消息数据。</td> 
  </tr> 
 </tbody> 
</table>

log文件存储内容：

## dump出来的具体日志数据内容
    Dumping /apps/Kafka/kafkalogs/kafka-loongshawn-01-0/00000000000022372103.log
    Starting offset: 22372103
    offset: 22372103 position: 0 CreateTime: 1532433067157 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 5d2697c5-d04a-4018-941d-881ac72ed9fd
    offset: 22372104 position: 0 CreateTime: 1532433067159 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 0ecaae7d-aba5-4dd5-90df-597c8b426b47
    offset: 22372105 position: 0 CreateTime: 1532433067159 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 87709dd9-596b-4cf4-80fa-d1609d1f2087
    ......
    ......
    offset: 22372444 position: 16365 CreateTime: 1532433067166 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 8d52ec65-88cf-4afd-adf1-e940ed9a8ff9
    offset: 22372445 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 5f5f6646-d0f5-4ad1-a257-4e3c38c74a92
    offset: 22372446 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 51dd1da4-053e-4507-9ef8-68ef09d18cca
    offset: 22372447 position: 16365 CreateTime: 1532433067168 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 80d50a8e-0098-4748-8171-fd22d6af3c9b
    ......
    ......
    offset: 22372785 position: 32730 CreateTime: 1532433067174 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: db80eb79-8250-42e2-ad26-1b6cfccb5c00
    offset: 22372786 position: 32730 CreateTime: 1532433067176 isvalid: true keysize: 4 valuesize: 36 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] key: 1 payload: 51d95ab0-ab0d-4530-b1d1-05eeb9a6ff00
    ......
    ......

从上可以看到 Kafka 消息的物理结构，其包含了 Kafka 消息的 offset 信息、Kafka 消息的大小信息、版本号等等。有了这些信息之后，我们就可以正确地读取到 Kafka 消息的实际内容。

## 如何查找 message ##

前面我们分析了 Kafka 的整套文件存储机制，也讲了如何定位、读取到 Kafka 消息的内容。那么我们现在就来模拟一下如何根据 offset 寻找到 Kafka 消息内容。

### 依据偏移量定位到文件 ###

例如我们要读取 Topic 为 kafka-loongshawn-01、Partition 为 1，并且 offset 为 22372118 的 Kafka 消息内容，我们应该怎么做呢？

*  定位数据文件夹  
    首先我们需要定位到具体 Partition 的数据文件夹，直接就是 Kafka 的数据存储目录，然后是以 Topic + Partition 命名的文件夹。即定位到kafka-loongshawn-01-1文件夹。

drwxr-x--- 2 root root 4096 Jul 26 19:35 kafka-loongshawn-01-0
    drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-loongshawn-01-1
    drwxr-x--- 2 root root 4096 Jul 24 20:15 kafka-loongshawn-01-2

*  定位索引文件  
    定位到数据文件夹后，我们可以将所有数据文件的文件名都列出来。依据偏移量22372118定位到索引文件为00000000000022372103.index

-rw-r----- 1 root root 512K Jul 24 19:51 00000000000000000000.index
    -rw-r----- 1 root root 1.0G Jul 24 19:51 00000000000000000000.log
    -rw-r----- 1 root root 768K Jul 24 19:51 00000000000000000000.timeindex
    -rw-r----- 1 root root 512K Jul 24 20:03 00000000000022372103.index
    -rw-r----- 1 root root 1.0G Jul 24 20:03 00000000000022372103.log
    -rw-r----- 1 root root 768K Jul 24 20:03 00000000000022372103.timeindex
    -rw-r----- 1 root root 512K Jul 24 20:15 00000000000044744987.index
    -rw-r----- 1 root root 1.0G Jul 24 20:15 00000000000044744987.log
    -rw-r----- 1 root root 767K Jul 24 20:15 00000000000044744987.timeindex
    -rw-r----- 1 root root  10M Jul 24 20:21 00000000000067117761.index
    -rw-r----- 1 root root 511M Jul 24 20:21 00000000000067117761.log
    -rw-r----- 1 root root  10M Jul 24 20:21 00000000000067117761.timeindex

关于如何定位到具体的索引文件的，依据索引文件的命名都是offset，且连续不重复，所以就可以知道每个索引文件的数据范围：

00000000000000000000.index -> 00000000 ~ 22372103
    00000000000022372103.index -> 22372104 ~ 44744987
    00000000000044744987.index -> 44744988 ~ 67117761

我们要寻找的是 offset 为 22372118 的消息，那其索引数据就存储在 00000000000022372103.index 这个索引文件中。

### 读取消息内容 ###

Kafka消息的索引文件

-rw-r----- 1 root root 512K Jul 24 20:03 00000000000022372103.index
    -rw-r----- 1 root root 1.0G Jul 24 20:03 00000000000022372103.log
    -rw-r----- 1 root root 768K Jul 24 20:03 00000000000022372103.timeindex

已经找到索引文件，我们要读取的消息为 22372118，那么这条消息在索引文件中是第 15 条消息。第 15 条在数据文件的物理位置为 7211。

根据上面的分析，我们知道：offset 为 22372118 的消息，其索引数据存储在 00000000000022372103.index，消息内容存储在 00000000000022372103.log 文件偏移位置为 7211 的地方。

那接下来结合偏移量以及消息的物理结构，直接读取到 offset 为 22372118 的消息内容了。有关数据文件如何读取具体消息的内容，有兴趣的可以自行深入了解下。这里就不开展。

## 稀疏索引 ##

从下图的 Kafka 消息索引文件可以看出，索引文件中并不存储每条消息在数据文件的偏移地址。这是为什么呢？

### 如何理解 ###

但要理解这个问题，必须明白为什么 Kafka 要这么做。

因为 Kafka 的消息太多了，如果把所有消息的「offset，物理偏移量」信息都存入 index 文件的话，那么这个文件太大了，无法一次性存入内存。而如果无法一次性存入内容，就会导致需要多次去读取，但每次去读磁盘又会降低读取效率。

对于消息量巨大的系统，比如几十亿级、百亿级，如果全部都存储，将会占用大量的磁盘空间和内存。反之若只存储1/2，1/5等，将会大大减少文件大小。

如存储 10 万个消息，但现在我只存 5 万个索引。Kafka 索引文件一样跳着存储。如果我要寻找第 8 条消息，那我只需要找到第 7 条消息的物理偏移量。之后在读取数据文件的时候直接跳过 1 条消息就可以了。

减少索引文件的大小，让内存可以全部读取进去。所有索引数据都存在内存中，减少磁盘 IO 的次数，极大地提高查找效率。这种索引存储方法叫：稀疏索引存储方式。

### 极端情况 ###

10万个消息，减少到了 100 个。那我去读取数据文件的时候，不就是一条一条的去判断了？

没错！极端情况下确实会出现这样一种问题。

### 密度和均匀 ###

所以稀疏索引需要保证两个很重要的点，一个是索引的密度，一个是索引的均匀程度。

*  索引的密度。 虽然减少索引的信息可以减少文件大小，但是会导致后续查找的时间成本。所以为了后续的查找方便，你不能减少太多索引信息。例如原本有 10 万个消息 offset 索引，你现在减到了 100 个，那就太少了。
 *  索引的均匀程度。 还是 10 万个消息的 offset 索引为例，如果你弄了 80000 个索引。这时候索引密度应该是没问题的。但是这 80000 个索引全都是第 1 - 80000 条消息的索引，最后一个索引是 80000 - 10 万的。那么当要读取的 offset 坐落于这个区间时，读取速度也会很慢。所以索引的分布也要均匀！

索引的多少与密度肯定有一个最佳平衡点，而我相信 Kafka 应该找到了这样一个平衡点。

## 总结 ##

本文重点了解以下几个点，科普个概念即可：

*  索引文件长啥样
 *  数据文件长啥样
 *  各自是如何存储
 *  稀松索引的理解

不过要留点尾巴，后续再开展学习下，比如：

*  kafka的快要如何保证，读写快如何理解？
 *  mmap如何理解？

## 引用 ##

1、[Kafka文件存储机制那些事][Kafka]  
2、[Apache Kafka® 是 一个分布式流处理平台. 这到底意味着什么呢?][Apache Kafka_ _ _.]  
3、[Kafka之数据存储][Kafka 1]  
4、[Kafka 凭什么速度那么快？][Kafka 2]  
5、[Kafka架构图][Kafka 3]

[95bfc2f6a9b3481c952ae64233d9a620.png_pic_center]: /images/20220829/dd7863c0d9c2433fa24c5ad2e2a3e60b.png
[Kafka]: https://tech.meituan.com/2015/01/13/kafka-fs-design-theory.html
[Apache Kafka_ _ _.]: https://kafka.apachecn.org/intro.html
[Kafka 1]: https://matt33.com/2016/03/08/kafka-store/
[Kafka 2]: https://www.infoq.cn/article/ukoqjkuwr0v0cs7u6p8o
[Kafka 3]: https://zhuanlan.zhihu.com/p/38269875