Kafka 提供两种日志清理策略：日志清理(Log Delete)与日志压缩(Log Compaction)。

1 日志清理

通过broker端参数log.cleanup.policy来设置日志清理策略，默认值为“delete”。如果要采用日志压缩的清理策略，则设置为“compact”。可以同时支持设置两种策略。粒度可以控制到主题级别。

1.1 日志删除

Kafka日志管理器有一个专门的日志删除任务来周期性地检测和删除不符合保留条件的日志分段文件。周期通过broker端参数log.retention.check.interval.ms来配置，默认值为5分值。

有3种保留策略：基于时间、基于日志大小、基于日志起始偏移量。

1.1.1 删除操作

将不符合保留条件的日志分段加入到deletableSegments集合中，如果其数量等于该日志文件中所有日志分段的数量，但该日志文件必须保证有一个活跃的日志分段，此时，会先切分出一个新的日志分段作为activeSegment，然后执行删除操作。

删除日志分段时，首先会从Logo对象所维护的日志分段跳跃表中移除待删除的日志片段，以保证没有线程对这些日志分段进行读取操作，然后将日志分段所对应的所有文件添加上”.deleted”的后缀，最后交由一个以“delete-file”命名的延迟任务来删除这些文件。

1.1.2 基于时间

日志删除任务会检查当前日志文件中是否有保留时间超过设定的阈值。阈值通过broker端参数log.retention.hours、log.retention.monutes和log.rention.ms来配置，默认值为7天。

检查日志的保留时间，是根据日志分段中最大的时间戳来计算。如果不存在，则取该日志最近修改的时间。

1.1.3 基于日志大小

删除任务会检查当前日志的大小是否超过设定的阈值（通过broker端参数log.retention.bytes来配置）。默认值为-1，表示无穷大。删除算法如下：

1. 计算日志文件的总大小和阈值的差值。
2. 从日志的第一个日志分段开始，将日志分段加入到deletableSegments中，直到集合中分段的总大小大等于阈值。
3. 删除这些日志分段。

1.1.4 基于日志起始偏移量

判断某日志分段的下一个日志分段起始偏移量baseOffset是否小等于logStartOffset（可以通过DeleteRecordsRequest请求指定）。如果是，则将这个日志分段加入到deletableSegments。

1.2 日志压缩

对于有相同key的不同value值，只保留最后一个版本。

如果应用只关心key对应的最新value，则可以开启Kafka日志压缩清理功能。

日志压缩会生成新的日志分段文件，因为其执行过后的偏移量不再是连续的，但这并不影响日志的查询。

1.2.1 清理检查点文件

在日志目录下，名为“cleaner-offset-checkpoint”的文件。

用于存储每个日志的最后清理偏移量（标识了日志清理线程已经清理到的位置）。 使得Kafka能够在系统重启或故障恢复后，从上次清理结束的位置继续清理日志。

图 清理检查点

起始节点：日志消息开始的偏移量（不一定从0 开始）。

checkpoint: 清理检查点，表示上次清理到的位置。

clean部分：已经清理过的部分，偏移量是断续的。 [起始节点,checkpoint)。

dirty部分：待清理的部分。[checkpoint,uncleanableOffset)。

为了避免当前活跃的日志分段activeSegment成为热点文件，其不会参与到清理中。

浑浊率 = dirty部分的日志大小 / （clean 部分的日志大小 + dirty部分的日志大小）

每个broker会启动日志清理线程负责清理任务，会选择“浑浊率”最高的日志文件进行清理。

1.2.2 清理过程

清理线程会使用“SkimpyOffsetMap”的对象来构建key与offset的映射关系的哈希表。最开始遍历dirty部分，把每个key的哈希值和最后出现的offset都保存在该Map中，第二次遍历clean + dirty部分，会检查每个消息是否复合条件（消息的offset是否大等于其key保存在map中的offset），不符合的会被清理。

注意：两个不同的key但哈希值相同的情况可能会发送，但概率非常小。

1.2.3 墓碑消息

消息的key不为null,但其value为null，那么此消息就是墓碑消息。清理线程发现墓碑消息会先进行上面常规的清理，并且保留墓碑消息一段时间，然后再删除。

1.2.4 日志片段合并

Log Compaction 执行过后的日志分段的大小会比原先的日志分段要小，为了防止出现太多的小文件，在实际的清理过程中并不对单个的日志分段进行单独清理，而是将日志文件从起始到uncleanableOffset的所有日志分段进行分组，同一组的多个日志分段清理过后，只会生成一个新的日志分段。

分组策略：按照日志分段的顺序遍历，每组中日志分段总大小不能超过segmentSize,且对应的索引文件总大小不能超过maxIndexSize。

图 执行Log Compaction

清理过程中，会将每个日志分组中需要保留的消息复制到以“.clean”为后缀的临时文件中，例如第一个分组中起始偏移量为0，则命名为00000000000000000000.log.clean。清理完成后将“.clean”修改为“.swap”，然后把原本的日志文件删除，最后才把“.swap”后缀去掉。索引文件的变换也是如此。