Kafka日志压缩需显式配置broker的log.cleaner.enable=true和topic的cleanup.policy=compact，消息必须带非空key，Producer发送时用稳定key（如订单ID）并确保value为最新状态，配合min.cleanable.dirty.ratio、delete.retention.ms等参数调优以保障压缩及时有效。
Java 中启用 Kafka 的日志压缩（Compaction）策略，核心不是“写代码删数据”，而是通过配置驱动 Broker 和 Topic 行为，让 Kafka 自动按 Key 保留最新值。它适用于状态类数据（如用户资料、设备配置、订单状态），而非事件流。
明确启用压缩并确保基础条件满足
Log Compaction 不是默认开启的，必须显式配置且依赖多个开关协同生效：
-
Broker 级必须开启 cleaner：确保
log.cleaner.enable=true（Kafka 默认为 true，但生产环境建议显式确认） -
Topic 级指定策略：设置
cleanup.policy=compact，不能只靠全局默认；若还需配合时间清理，可设为delete,compact - 消息必须带非空 Key：只有 key 不为 null 的消息才会参与压缩；key 为 null 的消息会被丢弃（除非同时启用了 delete 策略）
在 Java 应用中正确发送支持压缩的消息
Producer 发送时需保证 key 有意义且稳定（如用户 ID、订单号），value 包含当前完整状态：
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// Key 是订单ID，Value 是最新订单状态JSON
producer.send(new ProducerRecord<>("order-status", "ORD-1001", "{\"status\":\"shipped\",\"ts\":1723576800}"));
producer.send(new ProducerRecord<>("order-status", "ORD-1001", "{\"status\":\"delivered\",\"ts\":1723580400}"));
注意：相同 key 的多条消息写入后，压缩完成时仅保留最后一条（offset 最大的那条）。
立即学习“Java免费学习笔记（深入）”；
合理配置压缩相关参数提升稳定性
仅设 cleanup.policy=compact 不够，还需调优几个关键参数防止压缩滞后或失效：
-
min.cleanable.dirty.ratio（默认 0.5）：表示“脏数据”（未压缩部分）占比超过该值才触发压缩。流量低时可适当调低（如 0.1），避免长期不压缩 -
delete.retention.ms（默认 86400000，即 1 天）：控制 tombstone 消息（key 存在但 value 为 null）的保留时长，用于彻底删除某个 key。业务需清理某实体时，发一条key="ORD-1001", value=null即可 -
segment.ms或segment.bytes：影响段滚动频率，间接影响压缩粒度。建议保持segment.bytes=1GB左右，避免段过小导致元数据膨胀
主题创建与动态调整的推荐方式
不要依赖集群默认策略，应按业务语义为每个 Topic 单独配置：
-
新建 Topic 时指定：
kafka-topics.sh --create \ --topic user-profile \ --bootstrap-server localhost:9092 \ --config cleanup.policy=compact \ --config min.cleanable.dirty.ratio=0.1 \ --config delete.retention.ms=86400000 -
已有 Topic 动态修改：
kafka-configs.sh --alter \ --entity-type topics \ --entity-name device-config \ --add-config cleanup.policy=compact \ --bootstrap-server localhost:9092
压缩策略生效后，可通过 kafka-run-class.sh kafka.tools.DumpLogSegments 查看段内 key 分布，或监控 JMX 指标 kafka.log:type=LogCleanerStats 中的 cleaning-rate 和 dirty-bytes-ratio 来验证运行状态。
