Kafka清理策略配置示例

下面给一个比较完整的 Kafka compact 配置示例，包括 Broker 默认配置、Topic 配置以及生产环境常见组合。

1. Broker 端开启 LogCleaner

server.properties

# 开启日志清理线程
log.cleaner.enable=true

# LogCleaner线程数量
log.cleaner.threads=2

# Cleaner使用的buffer大小
log.cleaner.dedupe.buffer.size=134217728

# 触发compact的最小dirty比例
log.cleaner.min.cleanable.ratio=0.5

# Tombstone消息保留时间
log.cleaner.delete.retention.ms=86400000

说明：

2. 创建一个 compact Topic

例如创建用户状态 Topic：

kafka-topics.sh \
--create \
--bootstrap-server localhost:9092 \
--topic user-profile \
--partitions 6 \
--replication-factor 3

默认情况下：

cleanup.policy=delete

所以需要修改：

kafka-configs.sh \
--bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-profile \
--alter \
--add-config cleanup.policy=compact

3. 查看 Topic 配置

kafka-configs.sh \
--bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-profile \
--describe

看到：

cleanup.policy=compact

说明生效。

4. 推荐生产环境配置：compact + delete

通常状态 Topic 不希望无限增长。

例如：

用户状态：

user-profile

配置：

kafka-configs.sh \
--bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-profile \
--alter \
--add-config cleanup.policy=compact,delete

再设置保留时间：

kafka-configs.sh \
--bootstrap-server localhost:9092 \
--entity-type topics \
--entity-name user-profile \
--alter \
--add-config retention.ms=2592000000

含义：

30天后删除旧segment
+
同一个key只保留最新值

5. 一个完整 Topic 配置示例

假设：

Topic:
user-profile

Partition:
6

用途:
保存用户最新状态

配置：

cleanup.policy=compact,delete

# 最大保存30天
retention.ms=2592000000

# 每个partition最大10GB
retention.bytes=10737418240

# segment大小
segment.bytes=1073741824

# tombstone保留24小时
delete.retention.ms=86400000

# dirty达到40%触发compact
min.cleanable.dirty.ratio=0.4

对应效果：

user-profile

partition-0

segment1:
 user1=Tom(old)
 user2=Jack

segment2:
 user1=Tom(new)
 user3=Lucy


compact后:

segment:
 user1=Tom(new)
 user2=Jack
 user3=Lucy


30天后:
删除过期segment

6. Producer 示例

compact Topic 必须有 key。

正确：

ProducerRecord<String, String> record =
    new ProducerRecord<>(
        "user-profile",
        "user001",
        "{\"name\":\"Tom\",\"age\":20}"
    );

producer.send(record);

Kafka：

key=user001
       |
       hash
       |
       v
partition-2

之后更新：

new ProducerRecord<>(
    "user-profile",
    "user001",
    "{\"name\":\"Tom\",\"age\":21}"
);

Partition：

partition-2

offset 100:
key=user001
age=20

offset 200:
key=user001
age=21

compact 后：

offset 200:
key=user001
age=21

7. 删除一个 Key

发送 tombstone：

ProducerRecord<String, String> record =
    new ProducerRecord<>(
        "user-profile",
        "user001",
        null
    );

producer.send(record);

Kafka：

offset 300:
key=user001
value=null

compact 后：

user001 被删除

8. 实际生产推荐

状态类 Topic

例如：

• 用户信息
• 配置
• 商品库存

推荐：

cleanup.policy=compact,delete
min.cleanable.dirty.ratio=0.3
retention.ms=604800000

即：

保留最新状态
+
7天后清理无效历史

事件类 Topic

例如：

• 订单流水
• 支付记录
• 操作日志

不要 compact：

cleanup.policy=delete

因为：

创建订单
支付
发货
退款

这些历史事件不能丢。

一个比较典型的 Kafka 状态 Topic 配置就是：

cleanup.policy=compact,delete
min.cleanable.dirty.ratio=0.5
delete.retention.ms=86400000
retention.ms=2592000000
segment.bytes=1073741824

这也是很多生产环境保存“最新状态数据”的常用模式。