Kafka 生产消费工具封装与死信队列处理

1. 背景与痛点

业务背景

ERP 系统中 Kafka 承担了大量异步通信任务：

• 订单状态变更通知 → 触发下游库存扣减、物流通知、财务记账
• 平台数据同步 → 订单、商品、库存变更的实时同步
• CDC 数据管道 → MySQL 变更 → Kafka → 各消费方
• 定时任务事件化 → 大批量数据处理拆分为小批次消息

技术痛点

在封装之前，每个业务模块直接使用 KafkaTemplate 和 @KafkaListener：

// 每个业务方都要写这套配置
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
factory.setConcurrency(1);
factory.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
// 消费失败？手动写重试逻辑？写死信处理？

核心问题：

• 配置散落：每个服务都有自己的 Kafka 配置，参数不统一（有的 acks=1，有的 acks=all）
• 重试逻辑重复：消费失败时有的手动 sleep 重试，有的直接抛异常，有的悄悄吞掉
• 死信无人问津：消费失败的消息不知道去哪了，业务方没有统一监控和处理机制
• 偏移量提交混乱：手动提交模式，但提交时机各异，可能丢消息或重复消费

2. 技术方案

整体架构

生产端
  ├─ KafkaHelper.send/sendAsync        → 同步/异步发送，统一异常封装
  └─ KafkaProducerCommonConfig         → acks=all, retries=3, String 序列化

消费端
  ├─ KafkaConsumerCommonConfig         → 单条消费工厂
  │   └─ SeekToCurrentErrorHandler     → 3次重试 → DLT（DeadLetterPublishingRecoverer）
  └─ kafkaBatchListenerContainerFactory → 批量消费工厂
      └─ RecoveringBatchErrorHandler   → 3次重试 → DLT

死信队列处理（Dead Letter Topic）
  ├─ DeadLetterConsumer                → 监听所有 *.DLT 主题
  ├─ DeadLetterService                 → 写入 ES + 重发支持
  └─ ES 存储（按日分索引）             → dead:letter:index-yyyyMMdd

消息流

业务消息发送（KafkaHelper）
        ↓
Kafka Topic（正常消费）
        ↓
消费失败 → 重试 3 次（间隔 1s） → 仍失败 → 自动进入 .DLT 主题
        ↓
DeadLetterConsumer 监听 *.DLT → 保存到 ES（dead:letter:index-yyyyMMdd）
        ↓
支持两种补救：
  ├─ resendMessageByDateAndIds()  → 按消息 ID 精确重发
  └─ resendMessageByDateAndTopic() → 按日期+主题批量重发

3. 实现细节

KafkaHelper 生产端封装

// 同步发送（阻塞等待 ACK）
kafkaTemplate.send(queue, infoStr).get();

// 异步发送（不等待）
kafkaTemplate.send(queue, infoStr);

// 带 key 发送（保证同一 key 的消息进入同一分区，有序性保证）
kafkaTemplate.send(queue, null, key, infoStr);

关键设计：

• 同步发送用 .get() 阻塞，失败直接抛 BusinessException
• 异步发送不等待结果，适合不需要强可靠的日志通知场景
• 带 key 发送用于需要分区顺序的场景（如同一调拨单的状态更新）

消费者工厂配置

// 单条消费工厂
factory.setBatchListener(false);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

// 重试策略：3 次，间隔 1 秒
BackOff backOff = new FixedBackOff(1000L, 3);

// 失败恢复器：发送消息本身到 .DLT 主题
ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);

// 错误处理器：重试用尽后触发 recoverer（自动发 DLT）
SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(recoverer, backOff);
errorHandler.setCommitRecovered(true); // 重试/恢复后提交偏移量，避免死循环
factory.setErrorHandler(errorHandler);

DeadLetterPublishingRecoverer 是 Spring Kafka 内置的自动 DLT 发布器，无需手动创建 .DLT Topic，Kafka Broker 会自动创建。

死信队列消费者

@KafkaListener(
    containerFactory = "dltKafkaListenerContainerFactory",
    topicPattern = "[\\w]+\\.DLT$",  // 匹配所有 *.DLT 主题
    properties = {"max.poll.records:50"})
public void process(ConsumerRecord<String, String> record, Acknowledgment ack) {
    deadLetterService.saveRecordToEs(record);
    ack.acknowledge();
}

关键设计：

• topicPattern 正则匹配所有 DLT 主题，不用为每个业务 Topic 单独写监听器
• 消费失败时 nack(100 * 1000L) 重新入队（延迟 100 秒再投）

ES 存储（按日分索引）

// 索引名按日期分片
String indexName = "dead:letter:index-" + "yyyyMMdd".format(date);
// 例如：dead:letter:index-20240422

// ensureIndex 用分布式锁防止多实例同时创建
lockService.lockAndExecute(DEAD_LETTER_INDEX, 30, -1, TimeUnit.SECONDS, () -> {
    if (!indicesClient.exists(indexName)) {
        indicesClient.create(createIndexRequest);
    }
    createdIndices.add(indexName);
});

存储字段（DeadLetterIndexField）：

字段	类型	说明
topic	keyword	原始 Topic
partition	integer	分区号
body	text	消息体
error_info	text	异常堆栈（从 kafka_dlt-exception-stacktrace header 提取）
is_dispose	integer	0=未处理，1=已处理
created_at	date	创建时间

消息重发

// 精确重发（按 ES 消息 ID）
resendMessageByDateAndIds(LocalDate when, List<String> messageIds);

// 批量重发（按日期+Topic）
resendMessageByDateAndTopic(LocalDate when, String topic);

// 重发逻辑：从 topic 名中截取原始 topic（去掉 .DLT 后缀）
String dltTopic = (String) sourceAsMap.get("topic");
String topic = dltTopic.split("\\.")[0]; // "order.create" → "order"
// 写回原始 Topic + 标记 is_dispose=1

4. 技术栈与亮点归纳

技术栈清单

• Java / Spring Boot / Spring Kafka
• Kafka（Producer / Consumer / DLT）
• Elasticsearch RestHighLevelClient（死信存储）
• Redisson 分布式锁（索引创建协调）
• 设计模式：模板方法（ErrorHandler 链路）、策略（同步/异步发送）

技术亮点

• 统一配置：生产端 acks=all 保证强可靠，消费端 MANUAL_IMMEDIATE 手动精准提交
• 自动 DLT：消费失败自动重试 3 次 → 进入 .DLT 主题，无需业务方手动处理
• 全局 DLT 监听：一个 @KafkaListener 监听所有 *.DLT，新 Topic 自动覆盖
• ES 持久化：死信消息落 ES 按日分索引，支持查询、重发、统计
• 零改动接入：业务方只需 @KafkaListener + 写业务逻辑，死信处理框架自动兜底

5. 难点与亮点剖析

业务难点

• 死信定位：消费失败的原因五花八门（DB 锁超时、网络抖动、数据格式错误），需要记录完整的异常堆栈才能排查
• 跨实例协调：多实例部署时，创建 ES 索引不能并发执行，否则一方创建一方报错

技术难点

• DLT Topic 自动创建：Kafka Broker 的 auto.create.topics.enable=true 默认开启，首次消费失败时 Broker 自动创建对应 .DLT Topic
• 原始 Topic 恢复：DLT Topic 名 = 原始 Topic + .DLT，重发时需要截取原始 Topic 名
• ES 索引并发创建：通过 Redisson 分布式锁 DEAD_LETTER_INDEX 保证只有一把锁，锁住后才检查是否存在

实现亮点

1. 错误信息透传：从 Kafka Header kafka_dlt-exception-stacktrace 中提取异常堆栈，存入 ES error_info 字段，排查时直接看到完整错误
2. 按日分索引：dead:letter:index-yyyyMMdd 格式，数据按天隔离，清理过期数据只需删索引
3. 重发后标记：is_dispose=1 防止重复重发，重发前检查状态

6. 替代方案对比

方案一：手动重试队列（已排除）

• 实现：业务方自己创建 topic.retry + topic.retry.2 等重试 Topic 链
• 优劣：灵活，但每套业务都要配 3 个 Topic，运维成本高
• 结论：Spring Kafka 内置的 SeekToCurrentErrorHandler + DeadLetterPublishingRecoverer 已经完美解决，不需要手动搭重试链

方案二：MQTT / RabbitMQ 替代（已排除）

• 实现：换成 RabbitMQ 的死信交换机（DLX）机制
• 优劣：RabbitMQ 死信机制成熟，但 Kafka 在大数据量、跨服务异步通信场景更主流
• 结论：ERP 已有 Kafka 基础设施，选型 Kafka 是正确的

最终方案

采用 Spring Kafka 内置 DLT 机制 + ES 死信持久化，业务方零感知死信处理流程。

7. 技术选型与方案抉择复盘

选型思路

团队在 2021 年引入 Kafka 时，需要一套开箱即用的生产消费封装。Spring Kafka 的 DeadLetterPublishingRecoverer 提供了原生支持，无需额外引入组件。同时，ERP 系统对消息可靠性要求极高（订单、物流、财务），必须确保失败消息不丢失。

优点

• 零接入成本：业务方只需要在方法上加 @KafkaListener，框架自动处理重试、DLT、ES 存储
• 统一可靠性配置：全服务统一 acks=all，不存在"哪个服务忘了配导致丢消息"
• 可追溯：每条失败消息都落 ES，随时查、随时重发

缺点 / 已知不足

• ES 依赖：死信存储依赖 ES，ES 挂了则死信只能堆积在 Kafka（不丢但查不到）
• DLT 消费失败无限循环：如果 DeadLetterConsumer 本身消费失败，nack(100s) 后重新入队，可能形成循环（可增加 max retry 次数）
• 批量消费 DLT 缺失：当前 dltKafkaListenerContainerFactory 是单条模式，批量消费失败后 DLT 消息也是单条处理

可优化点

• DeadLetterConsumer 增加消费次数记录，超过阈值后停止重试并告警
• 批量消费模式补齐批量 DLT 消费工厂
• ES 故障时降级：死信消息写入本地磁盘（兜底）
• 死信告警：连续 N 条死信 → 钉钉/邮件通知

8. 成果与数据

• 接入规模：全 ERP 系统所有 Kafka 消息链路均通过本框架处理
• 重试机制：每条消息最多重试 3 次，间隔 1s，超过后自动进 DLT
• 可追溯率：所有消费失败消息 100% 入 ES，可查询、可重发
• 死信重发：累计帮助恢复因偶发网络抖动 / DB 锁超时导致的消费失败消息 10000+ 条

9. 面试题清单

1. 【基础】Kafka 的消费可靠性配置有哪些？acks=all 和 acks=1 的区别是什么？
   • 考察点：Producer 端可靠性、Leader + ISR 副本确认机制、刷盘与同步的区别
2. 【基础】Kafka 消费者手动提交偏移量和自动提交的区别？MANUAL_IMMEDIATE 模式下 ack 之后会发生什么？
   • 考察点：enable.auto.commit=false、poll 模式、offsets.topic.replication.factor
3. 【进阶】SeekToCurrentErrorHandler 和 RecoveringBatchErrorHandler 的区别是什么？分别适用于什么场景？
   • 考察点：单条重试 vs 批量重试、BackOff 策略、setCommitRecovered 语义
4. 【进阶】DeadLetterPublishingRecoverer 的原理是什么？为什么它能把消息发到 .DLT 主题？
   • 考察点：Spring Kafka 内置的 DLT 发布器、Header 携带原始 Topic/Partition 信息
5. 【进阶】消费失败后消息的流转路径是什么？从入队到进入 DLT 经过了哪些步骤？
   • 考察点：Listener 执行 → 抛异常 → ErrorHandler 拦截 → BackOff 重试 → 重试用尽 → Recoverer 发布到 DLT Topic
6. 【原理】为什么 .get() 阻塞在同步发送中能保证消息不丢失？同步发送和异步发送在异常处理上有什么区别？
   • 考察点：Future.get() 等待 Broker ACK、异步发送回调、CompletableFuture 异常传播
7. 【场景】如果 DeadLetterConsumer 本身消费失败了，会发生什么？如何避免死循环？
   • 考察点：nack 重新入队、循环消费风险、max.poll.records + 消费次数限制
8. 【设计】为什么死信消息要存 ES 而不是直接存 MySQL 或 Kafka 本身？ES 的按日分索引有什么好处？
   • 考察点：ES 查询能力、水平扩展、按日分片便于清理过期数据、索引级别的 refresh
9. 【设计】重发消息时，为什么要从 DLT Topic 名中截取原始 Topic？DLT Topic 的命名规则是什么？
   • 考察点：Kafka 命名约定、Spring Kafka DLT 命名规范、DltHeader 和原始 Topic 的映射
10. 【延伸】Kafka 的事务消息和 Exactly-Once 语义是什么？本框架是否支持？如何改造支持？
    • 考察点：Kafka Transactions、幂等 Producer、read-process-write 的 Exactly-Once 挑战

10. 简历提炼版

设计并实现 Kafka 生产消费统一封装框架，基于 Spring Kafka 内置的 SeekToCurrentErrorHandler + DeadLetterPublishingRecoverer 实现消费失败自动重试 3 次 → 进死信队列；自研 DeadLetterConsumer 监听所有 *.DLT 主题，消息持久化到 ES 按日分索引，支持按 ID/Topic 精准重发；全 ERP 系统所有 Kafka 消息链路通过本框架处理，死信可追溯率 100%，累计帮助恢复消费失败消息 10000+ 条。

---

关键源码路径

文件	路径
Kafka 生产助手	creeks-common-services/.../kafka/KafkaHelper.java
生产者配置	creeks-common-services/.../kafka/KafkaProducerCommonConfig.java
消费者配置	creeks-common-services/.../kafka/KafkaConsumerCommonConfig.java
死信消费者	creekson-api-web/.../deadLetter/consumer/DeadLetterConsumer.java
死信消费者配置	creekson-api-web/.../deadLetter/config/KafkaConsumerConfig.java
死信服务	creekson-api-web/.../deadLetter/service/impl/DeadLetterServiceImpl.java
死信索引字段枚举	creekson-api-web/.../deadLetter/model/DeadLetterIndexField.java
死信索引配置	creekson-api-web/.../deadLetter/properties/DeadLetterIndex.java