【Kafka】原理分析：消息的文件存储机制

小咪咪 2022-12-21 06:03 327阅读 0赞

消息发送端发送消息到broker上以后，消息是如何持久化的呢？那么这篇我们就来看看kafka是如何存储消息的。

首先需要了解的是，kafka是使用**日志文件的方式来保存**生产者和发送者的消息，每条消息都有一个oﬀset值来表示它在分区中的偏移量。

Kafka中存储的一般都是海量的消息数据，为了避免日志文件过大，Log并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个**目录**，这个目录的命名规则是`<topic_name>_<partition_id>`（如test-0）

一个topic的多个partition在物理磁盘上的保存路径，路径保存在 `/tmp/kafka-logs/topic_partition`（注：不是存储在kafka目录下）

![在这里插入图片描述][20201115023212321.png_pic_center]  
目录中包含了.index、.log、.snapshot、.timeindex等文件，分成这些文件的主要原因是LogSegment机制

### LogSegment ###

kafka是通过分段的方式将Log分为多个LogSegment，LogSegment是一个逻辑上的概念，

*  LogSegment组成：一个 LogSegment对应磁盘上的一个日志文件和一个索引文件
    
     *  .log：日志文件，用来记录消息
     *  .index：索引文件，用来保存消息的索引，作用是记录具体行数的消息与partition中的实际offset
 *  LogSegment作用
    
    假设kafka以partition为最小存储单位，可以想象当kafka producer不断发送消息，必然会引起partition文件的无线扩张，这样对于消息文件的维护以及被消费的消息的清理带来非常大的挑战，所以kafka以segment为单位又把partition进行细分
    
     *  可以减少每个日志文件的大小，每个partition相当于一个巨型文件被平均分配到多个大小相等的segment数据文件中（每个segment文件中的消息不一定相等）
     *  **方便已经被消费的消息的清理**，提高磁盘的利用率。（注：清理的消息即日志中的一条记录，而不是整个日志）

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

### 1.segment命名规则 ###

segment文件命名规则：

*  数值最大为64位long大小，20位数字字符长度，没有数字用0填充
 *  partion全局的第一个segment从0开始
 *  后续每个segment文件名为上一个 segment文件最后一条消息的oﬀset值进行递增。

假如第一个log文件的最后一个oﬀset为:5376，所以下一个segment的文件命名为: 00000000000000005376.log。对应的index为00000000000000005376.index

### 2.索引文件内容 ###

为了提高查找消息的性能，为每一个日志文件添加2个索引索引文件

*  索引文件分类
    
     *  OﬀsetIndex：对应.index文件
     *  TimeIndex：对应.timeindex, TimeIndex索引文件格式，它是映射时间戳和相对oﬀset
 *  索引文件中的内容
    
     *  index：当前消息在log文件的相对索引（从0开始），存储相对索引可以减小索引文件占用的空间
        
        举例，分段后的一个数据文件的offset是从20开始（文件名~~21.index），那么offset为25的Message在index文件中的相对offset就是25-20 = 5。
     *  position：物理偏移量，该消息在数据文件中的绝对位置，ByteBuﬀer的指针位置；只要打开文件并移动文件指针到这个position就可以读取对应Message
    
    注：Kafka采用了**稀疏索引**，即不会给所有的消息都建立索引，只会将日志文件的部分记录存储在索引文件中。

以\[4053,80899\]为例，在log文件中，对应的是第4053条记录，（position）为80899.

### 3.日志文件内容（消息内容） ###

通过kafka提供的命令，可以查看二进制的日志文件信息

sh kafka-run-class.sh kafka.tools.DumpLogSegments 
    --files /tmp/kafka-logs/test-0/00000000000000000000.log # 指定日志文件
    --print-data-log

一条消息，会包含很多的字段，如下：

offset: 5371 position: 102124 CreateTime: 1531477349286 isvalid: true keysize: -1 valuesize: 12
     magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false 
     headerKeys: [] payload: message_5371

*  oﬀset：分区偏移量，可能因为多分区导致不连续
 *  position：当前消息在文件中的定位标识，如行数
 *  createTime：创建时间
 *  keysize：key的大小，指的是Producer在发送消息时指定的key（非必需），是分区依据
 *  valuesize：value的大小、
 *  compresscodec：压缩编码
 *  payload：消息的具体内容

### 4.索引与消息对应关系 ###

index与log对应关系如下：

![image][]

*  文件名：当前索引文件和日志文件在整个Partition中的起始偏移量
 *  index：索引文件中的index对应日志文件中的是第几条记录。比如Message368772就可以从索引文件index=497查到偏移量为3，即在相应日志文件第三行。
    
    > => 消息在Partition的偏移量 = 文件名+index （注：由于稀疏索引，索引文件中只会有部分消息的索引）
 *  position：当前index在文件中的绝对位置，用于直接定位到消息

### 5.查找message过程 ###

在partition中如何通过oﬀset查找message？查找的算法是：

1.  索引文件：根据oﬀset的值，查找segment段中的index索引文件。
    
    注：由于索引文件命名是以上一个文件的最后一个oﬀset进行命名的，所以，使用二分查找算法能够根据oﬀset快速定位到指定的索引文件。
2.  近似position：找到索引文件后，根据oﬀset进行定位，找到索引文件中的符合范围的索引。
    
    注：由于是稀疏索引，所以可能只能定位到一条相近的消息
3.  具体消息：得到position以后，再到对应的log文件中，从position出开始查找oﬀset对应的消息，将每条消息的oﬀset与目标oﬀset进行比较，直到找到消息

示例：我们要查找oﬀset=368776这条消息（参照上面的图）

1.  那么先找到00000000000000368769.index,
2.  在索引文件中，找到索引6（对应的offset=368775）的具体位置position=1407（注：这次只能找到近似offset）
3.  到log文件中，根据1407这个position开始查找，比较每条消息的oﬀset是否大于等于368776。最后查找到对应的消息以后返回

### 6.日志清理策略 ###

前面提到过，日志的分段存储，一方面能够减少单个文件内容的大小，另一方面，方便kafka进行日志清理。日志的清理策略有两个（注：清理的是消息，即日志的一条记录，而不是删除整个日志）

*  根据消息的保留时间
    
     *  当消息在kafka中保存的时间超过了指定的时间，就会触发清理过程
     *  通过`log.retention.hours`进行设置，默认的保留时间是：7天
 *  根据topic存储的数据大小
    
     *  当topic所占的日志文件大小大于一定的阀值，则可以开始删除最旧的消息
     *  通过`log.retention.bytes`进行设置

kafka会启动一个后台线程，定期检查是否存在可以删除的消息。当其中任意一个达到要求，都会执行删除。

> 所以，并不是consumer消费完后，就删除相应的消息，而是根据具体的broker清理策略进行消息清理。这也意味着一条消息可以被不同consumer group的消费者重复消费。

### 7.日志压缩策略 ###

Kafka还提供了“日志压缩（Log Compaction）”功能，通过这个功能可以有效的减少日志文件的大小， 缓解磁盘紧张的情况，在很多实际场景中，消息的key和value的值之间的对应关系是不断变化的，就像数据库中的数据会不断被修改一样，消费者只关心key对应的最新的value。

因此，可以开启kafka 的日志压缩功能，服务端会在后台启动启动Cleaner线程池，定期将相同的key进行合并，只保留最新的 value值。日志的压缩原理是

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]

[20201115023212321.png_pic_center]: /images/20221120/bae5120c7c48493aa6635b7155e07b1c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/a261fcb6841c46809166226470bc48c5.png
[image]: /images/20221120/d8172623aaf242f688562bc8e41e4ab7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/92bf44ec2f3d4162b95eee879f092d7c.png