第 12 章 生产者与消费者怎样写得稳

生产者的五个关注点

第一，序列化：Kafka 保存字节，字符串和 JSON 需要编码；双方要约定字符集和字段格式。第二，Key 选择：它决定分区关联和热点风险。第三，可靠性：明确确认方式、幂等、超时与错误处理。第四，吞吐：批量与压缩能提高效率，但要评估延迟和 CPU。第五，关闭：正常退出时等待已发送任务完成，并处理失败。

batch.size 控制批次大小上限，linger.ms 影响等待聚合的时间，compression.type 选择压缩方式。它们需要用真实消息大小和流量验证，不是值越大越好。Kafka 4.0 起 Java 生产者的默认 linger.ms 调整为 5 毫秒，旧文章里“默认是 0”的结论不能直接套用到本书版本。

消费者的五个关注点

第一，配置正确的 group.id。第二，明确起点，auto.offset.reset=earliest 仅在没有有效进度时生效，不能覆盖已有书签。第三，决定提交时机。第四，控制批次处理时间，避免长时间不调用 poll 而触发组管理问题。第五，处理退出、再均衡和重试。

手动提交不是“关掉自动提交就万事大吉”。如果先把一批记录扔给线程池，马上提交整批进度，任务还没完成就宕机，仍会漏掉业务处理。

伪代码：至少一次处理

下面只是单线程、同步处理的思路，不是可以直接运行的语言代码：

关闭自动提交
循环：
    records = poll()
    按分区顺序处理本批记录
    每条记录：
        通过数据库事务与事件 ID 做幂等业务处理
        如果失败：进入既定重试策略，不跨过未完成记录提交
    本批各分区全部成功后：
        提交各分区“最后成功记录 offset + 1”

若批次中途失败，已成功但未提交的记录会重来，需要幂等。更细粒度提交可以降低重复范围，但会增加复杂度和开销。先把正确性做对，再优化。

数据格式演进

第一版有 order_id，第二版新增可选字段 coupon_id，通常比直接删除或改名旧字段容易兼容。消费者应明确必需字段、默认值、未知字段处理和错误策略。金额不能今天用元、明天无提示改成分。

Avro、Protobuf、JSON Schema 和 Schema Registry 是常见进阶工具；Schema Registry 并不是 Kafka Broker 自动附带的功能，兼容策略也需要团队约定。