【大数据】Kafka 数据存储-Toy模板网

这篇具有很好参考价值的文章主要介绍了【大数据】Kafka 数据存储。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1.文件目录

Kafka 中的消息是存储在磁盘上的，一个分区副本对应一个日志（Log）。为了防止 Log 过大，Kafka 又引入了 日志分段（LogSegment）的概念，将 Log 切分为多个 LogSegment ，相当于一个巨型文件被平均分配为多个相对较小的文件，这样也便于消息的维护和清理。事实上，Log 和 LogSegment 也不是纯粹物理意义上的概念，Log 在物理上只以文件夹的形式存储，而每个 LogSegment 对应于磁盘上的一个日志文件和两个索引文件，以及可能的其他文件（比如以 .txnindex 为后缀的事务索引文件），下图为 Topic、Partition、副本、Log 和 LogSegment 之间的关系。

kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

2.日志分段

虽然一个 Log 被拆为多个分段，但只有最后一个 LogSegment（当前活跃的日志分段）才能执行写入操作，在此之前所有的 LogSegment 都不能写入数据。当满足以下其中任一条件会创建新的 LogSegment。

当前日志分段文件的大小超过了 Broker 端参数 log.segment.bytes 配置的值，默认值为 $1073741824$ ，即 $1 GB$ 。
当前日志中第一条消息的时间戳与当前系统的时间戳的差值大于 log.roll.ms 或 log.roll.hours 参数配置的值。如果同时配置了 log.roll.ms 和 log.roll.hours，那么以 log.roll.ms 为准。默认只配置了 log.roll.hours 参数，其值为 $168$ ，即 $7$ 天。
偏移量索引文件或时间戳索引文件的大小达到 Broker 端参数 log.index.size.max.bytes 配置的值，默认值为 $10485760$ ，即 $10 M$ 。
追加的消息的偏移量与当前日志分段的偏移量之间的差值大于 Integer.MAX_VALUE，即（offset - baseOffset）> Integer.MAX_VALUE。

在索引文件切分的时候，Kafka 会关闭当前正在写入的索引文件并置为只读模式，同时以可读写的模式创建新的索引文件，默认大小为 $1 GB$ 。当下次索引切分时才会设置为实际大小。也就是说，之前的 Segment 都是实际大小，活跃的 Segment 大小为 $1 G$ 。

3.日志索引

索引的主要目的是提高查找的效率。

Kafka 采用 稀疏索引（sparse index）的方式构造消息的索引，它并不保证每个消息在索引文件中都有对应的索引项。而是每当写入一定量（由 Broker 端参数 log.index.interval.bytes 指定，默认 $4 K B$ ）的消息时，索引文件会增加一个索引项。

3.1 偏移量索引

一条偏移量索引包含两部分数据，如图：

kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

relativeOffset：相对偏移量，表示消息相对于 baseOffset 的偏移量，当前索引文件的文件名即为 baseOffset。
position：物理地址，也就是消息在日志分段文件中对应的物理位置。
baseOffset：Segment 第一个 Message 的 Offset。

消息查找过程
kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

如果我们要查找偏移 $23$ 的消息，那么应该怎么做呢？首先通过二分法在偏移量索引文件中找到不大于 $23$ 最大索引项，即 [22,656]，然后从日志分段文件中的物理位置 $656$ 开始顺序查找偏移 $23$ 的消息。

以上是比较简单的情况，如下图所示，如果要查找要查找偏移 $268$ 的消息，那么应该怎么办呢？

kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

首先肯定是定位到 baseOffset = 251 的日志分段，然后计算相对偏移量 relativeOffset 为 $268 - 251 = 17$ ，之后再在对应的索引文件中找到不大于 $17$ 的索引项，最后根据索引项中的 position 定位到具体的日志分段文件位置开始查找目标消息。

那么如何查找 baseOffset 25 的日志分段的呢？Kafka 使用了跳跃表的结构。Kafka 的每个日志对象中使用了 ConcurrentSkipListMap 来保存各个日志分段，每个日志分段的 baseOffset 作为 Key ，这样可以根据指定偏移量来快速定位到消息所在的日志分段。

3.2 时间戳索引

时间戳索引也是包含两部分数据，如图：

kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

timestamp：当前日志分段最大的时间戳。
relativeOffset：时间戳所对应的消息的相对偏移量，也就是偏移量索引中偏移量。

时间戳索引文件中包含若干时间戳索引项，每个追加的时间戳索引项中的 timestamp 必须大于之前追加的索引项的 timestamp，否则不予追加。

消息查找过程

kafka数据文件目录,# Kafka,大数据,kafka,消息队列,数据同步

如果要查找指定时间戳 targetTimeStamp = 1526384718288 开始的消息，首先是找到不小于指定时间戳的日志分段。这里就无法使用跳跃表来快速定位到相应的日志分段了，需要分以下几个步骤来完成。

targetTimeStamp 和每个日志分段中的最大时间戳对比，直到找到不小于 targetTimeStam 所对应的日志分段。（注：日志分段中的最大时间戳的计算是先查询该日志分段所对应的时间戳索引文件，找到最后一条索引项，若最后一条索引项的时间戳字段值大于 $0$ ，则取其值，否则取该日志分段的最近修改时间。）
找到相应的日志分段之后，在时间戳索引文件中使用二分查找算法查找到不大于 targetTimeStamp 最大索引项，即 [1526384718283, 28]，如此便找到了相对偏移量 $28$ 。
在偏移量索引文件中使用二分算法查找到不大于 $28$ 的最大索引工页，即 [26,838]。
从第一步中找到的日志分段文件中的 $838$ 的物理位置开始查找不小于 targetTimeStamp 的消息。