“ 要深入学习Kafka,理解Kafka的存储机制是非常重要的。本文介绍Kafka存储消息的格式以及数据文件和索引组织方式,以便更好的理解Kafka是如何工作的。”
Kafka消息存储格式
Kafka为了保证消息的可靠性,服务端会将接收的消息进行序列化并保存到磁盘上(Kafka的多副本存储机制),这里涉及到消息的存储格式,即消息编码后落到磁盘文件上的二进制的数据格式。下图是根据Kafka 3.0官方文档整理的消息格式:
包含三个部分:BatchRecords、Record,以及Header的编码格式。
BatchRecords是Kafka数据的存储单元,一个BatchRecords中包含多个Record(即我们通常说的一条消息)。BatchRecords中各个字段的含义如下:
字段名 | 含义 |
---|---|
baseOffset | 这批消息的起始Offset |
partitionLeaderEpoch | 用于Partition的Recover时保护数据的一致性,具体场景可以见KIP101 |
batchLength | BatchRecords的长度 |
magic | 魔数字段,可以用于拓展存储一些信息,当前3.0版本的magic是2 |
crc | crc校验码,包含从attributes开始到BatchRecords结束的数据的校验码 |
attributes | int16,其中bit0~2中包含了使用的压缩算法,bit3是timestampType,bit4表示是否失误,bit5表示是否是控制指令,bit6~15暂未使用 |
lastOffsetDelta | BatchRecords中最后一个Offset,是相对baseOffset的值 |
firstTimestamp | BatchRecords中最小的timestamp |
maxTimestamp | BatchRecords中最大的timestamp |
producerId | 发送端的唯一ID,用于做消息的幂等处理 |
producerEpoch | 发送端的Epoch,用于做消息的幂等处理 |
baseSequence | BatchRecords的序列号,用于做消息的幂等处理 |
records | 具体的消息内容 |
一个BatchRecords中可以包含多条消息,即上图中的Record,而每条消息又可以包含多个Header信息,Header是Key-Value形式的。Record和Header的格式都非常简单,就不对其中的字段做解释了。
Log Segment
在Kafka中,一个Topic会被分割成多个Partition,而Partition由多个更小的,称作Segment的元素组成。
Kafka一个Partition下会包含类似上图中的一些文件,由log、index、timeindex三个文件组成一个Segment,而文件名中的(0和35)表示的是一个Segment的起始Offset(Kafka会根据log.segment.bytes的配置来决定单个Segment文件(log)的大小,当写入数据达到这个大小时就会创建新的Segment)。log、index、timeindex中存储的都是二进制的数据(log中存储的就是上一部分介绍的BatchRecords的内容,而index和timeindex分别是一些索引信息。)
下图是log文件中数据解析后的示意图(也就是本文第一部分BatchRecords格式)。其中16开头的这一行表示一个第一条消息的Offset是16的BatchRecord,而24开头的这一行表示的是一个第一条消息的Offset是24的BatchRecord。
索引
我们知道Kafka中每个Consumer消费一个Partition都会有一个关联的Offset表示已经处理过的消息的位置。通常Consumer会根据Offset连续的处理消息。而通过Offset从存储层中获取消息大致分为两步:
其中第一步可以直接根据Segment的文件名进行查找(上面已经介绍了Segment的文件面就是它包含的数据的起始Offset)。第二步则需要一些索引信息来快速定位目标数据在Segment中的位置,否则就要读取整个Segment文件了,这里需要的索引信息就是上面的index文件存储的内容。
index文件中存储的是Offset和Position(Offset对应的消息在log文件中的偏移量)的对应关系,这样当有Offset时可以快速定位到Position读取BatchRecord,然后再从BatchRecord中获取某一条消息。比如上述Offset25会被定位到24这个BatchRecord,然后再从这个BatchRecord中取出第二个Record(24这个BatchRecord包含了24、25两个Record)。
注意,Kafka并不会为每个Record都保存一个索引,而是根据log.index.interval.bytes等配置构建稀疏的索引信息。
除了index索引文件保存Offset和Position的映射关系外,Kafka中还维护了timeindex,保存了Timestamp和Offset的关系,用于应对一些场景需要根据timestamp来定位消息。timeindex中的一个(timestampX,offsetY)元素的含义是所有创建时间大于timestampX的消息的Offset都大于offsetY。
同样的,timeindex也采用了稀疏索引的机制,使用和index相同的配置(log.index.interval.bytes),所以timeindex和index是一一对应的。
总结
本文首先介绍了Kafka消息的存储格式,然后介绍了Kafka是如何索引(index & timeindex)存储的数据的。看完索引部分后遗留了一个疑问:每次读取消息都要先根据索引读取Position信息,然后再根据Position去读数据,而索引又是稀疏索引(查找索引也是要开销的),这样效率是否会比较低呢?如果利用Consumer总是顺序读取消息的特性,每次读取数据时都带上一些上下文信息(比如上一次Offset对应的Position信息)直接去读Log数据效率是否会更高?欢迎留言交流!
本文分享自 MessageQueue 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!