Kafka 主题压缩机制详解

1. 概述

Apache Kafka 是一个强大的分布式事件流平台，广泛用于实时数据处理。其核心采用追加写日志（append-only log）结构，非常适合事件驱动架构。然而，由于日志持续增长，Kafka 必须提供日志清理策略来管理数据生命周期。

Kafka 提供了两种日志清理策略：compact（压缩） 和 delete（删除）。它们的行为不同，适用于不同的数据保留需求。本文重点介绍 compact 策略。

2. delete 策略的局限性

delete 策略是基于时间的日志清理机制。每条消息的生命周期受时间限制，超过设定的时间后，后台进程会将其删除以释放磁盘空间。

这种策略适用于短暂事件，比如应用日志、传感器数据或通知触发器。这些数据通常只处理一次，旧数据即使未被消费，其价值也远不如最新数据。

但有些场景下，我们不能使用 delete 策略。例如：

在事件溯源（Event Sourcing）系统中，Kafka 作为持久化事件存储，不能因为消息“过期”就被删除。
系统需要保留某个 key 的最新状态，以便未来重建状态。

这时就需要使用 compact 策略。

3. 日志压缩机制

compact 策略对每条消息进行基于 key 的保留处理。它会将相同 key 的消息视为重复数据，并在清理过程中保留每个 key 的最新一条消息，删除旧记录。

启用 compact 策略的 Kafka 主题被称为 compacted topic（压缩主题）。要启用该策略，只需设置主题的 cleanup.policy=compact。

3.1 compact 与 delete 策略的区别

delete 策略：仅基于消息的时间戳删除数据。
compact 策略：基于消息 key 删除重复数据，保留每个 key 的最新记录。

示例说明

假设一个 Kafka 主题 Bank.SavingsAccount，其分区 0 包含如下日志段：

offset 0: key=0001, value=...
offset 1: key=0002, value=...
offset 2: key=0001, value=...
offset 3: key=0001, value=...

使用 delete 策略（保留 2 天）：

offset 0 和 1 的消息会被删除（超过保留时间）。

使用 compact 策略：

offset 0 和 2 的消息会被删除（key=0001 的最新记录是 offset 3）
offset 1 的消息保留（key=0002 只有一条记录）

✅ 结论：compact 策略基于 key 删除旧记录，delete 策略基于时间删除旧消息。

3.2 分区结构

Kafka 的每个主题分区由多个日志段（log segments）组成，每个日志段是一个文件：

/kafka-logs/
└── Bank.SavingsAccount-0/
    ├── 00000000000000000000.index
    ├── 00000000000000000000.log
    ├── 00000000000000000000.timeindex
    ├── 00000000000000000150.index
    ├── 00000000000000000150.log
    ├── 00000000000000000150.timeindex
    ├── leader-epoch-checkpoint
    └── partition.metadata

.log 文件：存储实际消息。
.index 文件：记录 offset 到消息位置的映射。
.timeindex 文件：记录时间戳到 offset 的映射。

3.3 压缩机制详解

Kafka 启动一个后台线程 Log Cleaner 来执行压缩操作。

工作流程：

扫描所有启用 compact 策略的分区。
根据 min.cleanable.dirty.ratio（默认 0.5）判断是否需要压缩。
读取日志段，构建 key → 最新 offset 的映射。
重写日志段，跳过旧记录（即 key 的 offset < 映射中的最新 offset）。
生成新的 .log、.index 和 .timeindex 文件，替换旧文件。

⚠️ 注意：只有当消息存在时间超过 min.compaction.lag.ms 才会被压缩，防止刚写入的消息被误删。

3.4 监控压缩操作

压缩操作会带来 I/O 和 CPU 开销，需监控以下指标：

LogCleaner_cleaner_recopy_percent_Value：表示压缩过程中重写了多少日志。值越高，说明压缩效率越高。
LogCleaner_max_clean_time_secs_Value：最长压缩耗时。若耗时过长，可能需要增加线程数 log.cleaner.threads。

✅ 建议：定期检查日志压缩性能，避免影响 Kafka 整体稳定性。

3.5 常见踩坑点

使用随机生成的 message key 会导致压缩失效。

compact 策略依赖 key 来判断重复数据。
如果每个消息的 key 都是唯一的，就没有重复项可以被压缩。

❌ 错误示例：

String key = UUID.randomUUID().toString(); // ❌ 不适合 compact 策略
producer.send(new ProducerRecord<>("topic", key, value));

✅ 正确做法：

String key = "user-12345"; // ✅ 固定 key，便于压缩
producer.send(new ProducerRecord<>("topic", key, value));

4. 总结

Kafka 提供了两种日志清理策略：

delete：基于时间删除旧消息。
compact：基于 key 保留最新记录，删除重复数据。

compact 策略适用于需要保留 key 最新状态的场景，如事件溯源、状态快照等。使用时需注意：

合理设置 key，避免使用唯一值。
监控 Log Cleaner 的性能指标。
调整 min.cleanable.dirty.ratio 和 log.cleaner.threads 提升效率。

使用得当，compact 策略能显著减少磁盘占用，同时保留关键状态信息。

Persistence

REST

Security