kafka服务端之日志存储

日志布局

Ka饮a 中的消息是以主题为基本单位进行归类的， 各个主题在逻辑
上相互独立。 每个主题又可以分为一个或多个分区， 分区的数量可以在主题创建的时候指定，也可以在之后修改。 每条消息在发送的时候会根据分区规则被追加到指定的分区中， 分区中的每条消息都会被分配一个唯 一的序列号， 也就是通常所说的偏移量(offset )。

如果分区规则设置得合理， 那么所有的消息可以均匀地分布到不同的分区中， 这样就可以实现水平扩展。 不考虑多副本的情况， 一个分区对应一个日志(Log)。 为了防止Log过大，Ka住a又引入了日志分段(LogSegment)的概念，将Log切分为多个LogSegment, 相当于 一个巨型文件被平均分配为多个相对较小的文件， 这样也便于消息的维护和清理。 事实上， Log 和LogSegnient也不是纯粹物理意义上的概念， Log在物理上只以文件夹的形式存储， 而每个LogSegment对应于磁盘上的 一个日志文件 和两个索引文件， 以及可能的其他文件（比如以" . txnindex"为后缀的事务索引文件）。如下图简单描绘了日志存储结构。

向Log中追加 消息时是顺序写入的，只有最后 一 个LogSegment才能执行写入操作， 在此之 前所有的LogSegment都 不能写入数据。 为了方便描述， 我们将最后 一 个LogSegment称为"activeSegment" , 即表示当前活跃的日志分段。 随着消息的不断写入 ， 当activeSegment满足一 定 的条件 时 ， 就需要创建新的activeSegment, 之后追加的消息将写入新的activeSegment。

为了便于消息的检索， 每个LogSegment中的日志文件 （以 " .log"为文件后缀）都有对应的两个索引文件：偏移量 索引文件（以".index"为文件后缀）和时间戳索引文件（以".timeindex"为文件后缀） 。 每个LogSegment都有 一 个基准偏移量baseOffset, 用来表示当前LogSegment中第 一 条消息的offset。 偏移量是一 个64位的长整型数， 日志文件和两个索引文件都是根据 基准偏移量(baseOffset)命名 的， 名称固定为20 位数字， 没有达到的位数则用0填充。 比如第一 个LogSegment的基准偏移量为O, 对应的日志文件为00000000000000000000.log。

日志索引

每个日志分段文件对应了两个索引文件，主要用来提高查找消息的效率。偏移量索引文件用来建立消息偏移量（ offset ）到物理地址之间的映射关系，方便快速定位消息所在的物理文件位置；时间戳索引文件则根据指定的时间戳（ timestamp ）来查找对应的偏移量信息。

Kafka 中的索引文件以稀疏索引（ sparse index ）的方式构造消息的索引，它并不保证每个消息在索引文件中都有对应的索引 I页 。 每当写入一定量（由 broker 端参数 log.index.interval.bytes 指定，默认值为 4096 ，即 4KB ）的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项，增大或减小 log.index . interval.bytes的值，对应地可以增加或缩小索引项的密度。

稀疏索引通过 MappedByteBuffer 将索引文件映射到内存中，以加快索引的查询速度。偏移量索引文件中的偏移量是单调递增的，查询指定偏移量时，使用二分查找法来快速定位偏移量的位置，如果指定的偏移量不在索引文件中，则会返回小于指定偏移量的最大偏移量 。 时间戳索引文件中的时间戳也保持严格的单调递增，查询指定时间戳时，也根据二分查找法来查找不大于该时间戳的最大偏移量，至于要找到对应的物理文件位置还需要根据偏移量索引文件来进行再次定位。稀疏索引的方式是在磁盘空间、内存空间、查找时间等多方面之间的一个折中。

对非当前活跃的日志分段而言，其对应的索引文件内容己经固定而不需要再写入索引项，所以会被设定为只读 。 而对当前活跃的日志分段 C activeSegment ）而言，索引文件还会追加更多的索引项，所以被设定为可读写。在索引文件切分的时候， Kafka 会关闭当前正在写入的索引文件并置为只读模式，同时以可读写的模式创建新的索引文件，索引文件的大小由 broker 端参数 log . index.size . max.bytes 配置。 Kafka 在创建索引文件的时候会为其预分配log.index . size .m ax . bytes 大小的空间，注意这一点与日志分段文件不同，只有当索引文件进行切分的时候， Kafka 才会把该索引文件裁剪到实际的数据大小 。 也就是说，与当前活跃的日志分段对应的索引文件的大小固定为 log . index.s 工 ze.max.bytes ，而其余日志分段对应的索引文件的大小为实际的占用空间。

日志清理

Kafka 将 消息存储在磁盘中，为了 控制磁盘占用空间的不断增加就需要对消息做一 定的清理操作。 Kafka 中 每 一个分区副本都对应 一个 Log, 而Log又可以分为多个日志分段， 这样也便于日志的清理操作。 Kafka提供了两种日志清理策略。

• (1)日志删除(LogRetention): 按照 一 定的保留策略直接删除不符合条件的日志分段。
• (2)日志压缩 (Log Compaction): 针对每个消息的key进行整合， 对千有相同 key的不 同value 值， 只保留最后 一个版本。

我们可以通过broker端参数log.cleanup.police 来设置 日志清理策略，此参数的默认
值为"delete " , 即采用日志删除的清理策略。 如果要采用日志压缩的清理策略， 就需要将log.cleanup.police 设置为"compact" , 并且还需要将log.cleaner.enable (默认值
为true )设定为true。 通过将log.cleanup.police 参数 设置为"del ete,compact" , 还可以
同时支持日志删除和日志压缩两种策略。 日志清理的粒度可以控制到主题级别， 比如与log.cleanup.police 对应的主题级别的参数为cleanup.police。

日志删除

在Kafka 的日志管理器中会有一个专门的日志删除任务来周期性地检测和删除不符合 保留条件的日志分段文件，这个周期可以通过broker端参数log.retention.check.interval.ms来配置 ，默认值为300000, 即5分钟。 当前日志分段的保留策略有3 种：基于时间 的保留策略、基于日志大小的保留策略 和基于日志起始偏移量的保留策略 。实际应用中这些策略只要命中一个就可以触发日志删除。

基于时间

日志删除任务会检查当前日志文件中是否有保留时间超过设定的阙值(retentionMs)来寻找可删除的日志分段文件集合(deletableSegments)。etentionMs可以通过broker
端参数log.retention.hours、log.retention.minutes和log.retention.ms来配
置 ， 其 中 log.retention.ms 的优先级最高， log.retention.minutes 次之，
log.retention.hours最低。 默认情况下只配置 了log.retention.hours参数， 其值为
168, 故默认情况下日志分段文件的保留时间为7天。

查找过期的日志分段文件，并不是简单地根据日志分段的最近修改时间lastModifiedTime来计算的， 而是根据日志分段中最大的时间戳largestTimeStamp来计算的。 因为日志分段的lastModifiedTime可以被有意或无意地修改， 比如执行了touch操作 ， 或者分区副本进行了重新分配，lastModifiedTime并不能真实地反映出日志分段在磁盘的保留时间。要获取日志分段中的最大时间戳largestTimeStamp的值， 首先要查询该日志分段所对应的时间戳索引文件，查找 时间戳索引文件中最后 一条索引项， 若最后 一 条索引项的时间戳字段值 大于O,则取其值， 否则才设置为最近修改时间 lastModifedTime 。

若待删除的日志分段的总数等千该日志文件中所有的日志分段的数量， 那么说明所有的日志分段都已过期， 但该日志文件中还要有一个日志分段用千接收消息的写入， 即必须要 保证有一个活跃的日志分段activeSegment, 在此种情况下， 会先切分出一个新的日志分段作为activeSegment, 然后执行删除操作。

删除日志分段时，首先会从Log对象中所维护日志分段的跳跃表中移除待删除的日志分段，以保证没有线程对这些日志分段进行读取操作。 然后将日志分段所对应的所有文件添加上".deleted"的后缀（当然也包括对应的索引文件）。 最后交由 一个以"delete-fi le"命名的延迟任务来删 除这些 以 ".deleted "为 后 缀的 文 件 ， 这个 任务的 延 迟执 行 时 间可 以 通 过file.delete.delay.ms参数来调配， 此参数的默认值为60000, 即1 分钟。

基千日志大小

日志删除任务会检查当前日志的大小是否超过设定的阀值 (retentionSize)来寻找可删除的日志分段的文件集合(deletableSegments), 如图所示。retentionSize可以通过broker端参数log.retention.bytes来配置 ，默认值为-1 , 表示无穷大。注意log.retention.bytes
配置的是Log中所有日志文件的总大小， 而不是单个日志分段（确切地说应该为log 日志文件）的大小。 单个日志分段的大小由broker 端参数log.segment.bytes 来限制， 默认值为1073741824, 即1GB。

基于日志大小的保留策略与基于时间的保留策略类似， 首先计算日志文件的总大小 size 和retentionSize的差值 diff, 即计算需要删除的日志总大小， 然后从日志文件中的第 一个日志分段开始进行查找可删除的日志分段的文件集合deletableSegments。 查找出deletableSegments之后就执行删除操作， 这个删除操作和基千时间的保留策略的删除操作相同， 这里不再赘述。

基于日志起始偏移量

一般情况下， 日志文件的起始偏移量 logStartOffset 等于第 一个日志分段的baseOffset , 但这并不 是绝对的， logStartOffset 的值 可 以 通 过DeleteRecordsRequest请求、 日志的清理和截断等操作 进行修改。

基于日志起始偏移量的保留策略的判断依据是某日志分段的下一个日志分段的起始偏移量baseOffset 是否小于等于logStartOffset, 若是， 则可以删除此日志分段 。 如图所示， 假设logStartOffset等于25, 日志分段 l 的起始偏移量为O, 日志分段2的起始偏移量为11, 日志分段 3的起始偏移量为23, 通过如下动作收集可删除的日志分段的文件集合deletableSegments:

• (1 )从头开始遍历每个日志分段 ， 日志分段 l 的下一个日志分段的起始偏移量为11, 小 于logStartOffset的大小，将日志分段 l加入deletableSegments。
• (2)日志分段2的下 一个日志偏移量的起始偏移量为23 , 也小于logStartOffset的大小，将日志分段2页加入deletableSegments。
• (3)日志分段3的下 一个日志偏移量在logStartOffset的右侧， 故从日志分段3开始的所有日志分段都不会加入deletableSegments。

收集完可删除的日志分段的文件集合之后的删除操作同基于日志大小的保留策略和基千时间的保留策略相同， 这里不再赘述。

日志压缩

Kafa中的Log Compaction是指在默认的日志删除(Log Retention)规则之外提供的 一种
清理过时数据的方式。 如图所示， Log Compaction对于有相同key的不同value值， 只保留最后 一个版本。如果应用只关心key对应的最新value值，则可以开启Kafka的日志清理功能，Kafa会定期将相同key的消息进行合并， 只保留最新的value值。

Log Compaction执行前后， 日志分段中的每条消息的偏移量和写入时的偏移量保待 一致。Log Compaction 会生成新的日志分段 文件， 日志分段中每条消息的物理位置会重新按照新文件来组织。 Log Compaction执行过后的偏移量不再是连续的， 不过这并不影响日志的查询。

Log Compaction会保留key相应的最新value值 ，那么当需要删除 一个key时怎么办? Kafka提供了 一个墓碑消息(tombstone)的概念， 如果 一 条消息的key不为null, 但是其value为null,那么此消息就是墓碑消息 。 日志清理线程发现墓碑消息时会先进行常规的清理， 并保留墓碑消息 一 段 时间。

Log Compaction执行过后的日志分段的大小会比原先的日志分段的要小， 为了防止出现太多的小 文件， Kafa在实际清理过程 中并不对单个的日志分段进行单独清理，而是将日志文件中offset从0至frstUncleanableOffset的所有日志分段进行分组，每个日志分段只属于一 组，分组策略为：按照日志分段的顺序遍历， 每 组中日志分段的占用空间大小之和不超过segmentSize（可以通过broker端参数log.segment.bytes设置， 默认值为1GB) , 且对应的索引文件占用大小之和不超过maxindexSize(可以通过broker端参数log.index.interval.bytes设置， 默认值为10MB) 。 同 一个组的多个日志分段清理过后， 只会生成 一个新的日志分段 。

总结

注意将 日志压缩和日志删除区分开， 日志删除是指清除整个日志分段， 而日志压缩是针对相同key的消息的合并清理 。