Kafka：消费者重试与死信队列的对应模式分析

在 Spring Kafka 中，消费者重试（@RetryableTopic） 与 死信队列（@DltHandler） 的对应关系及 死信 Topic 名称拼接 是保障消息可靠处理的核心机制。本文将详细解析两者的对应模式、Topic 名称生成规则，并结合生产实践给出最佳实践。

一、`@RetryableTopic` 与 `@DltHandler` 的对应关系

Spring Kafka 通过 “隐式绑定+显式隔离” 实现重试与死信的关联，核心原则是 “一对一绑定”（一个重试方法对应一个死信处理方法），具体规则如下：

1. 基础对应规则（默认模式）

要素	规则
所在类	@DltHandler 方法必须与 @RetryableTopic 方法在同一个类中（Spring 容器通过类作用域查找）。
绑定方式	默认绑定到最近定义的 @RetryableTopic 方法（若类中只有一个 @RetryableTopic，则直接绑定）。
参数兼容性	@DltHandler 方法的参数需与原始消息类型兼容（如原始消息是 Order 对象，@DltHandler 可直接接收 Order，或通过 ConsumerRecord 获取完整上下文）。

2. 多方法冲突与解决方案

若同一类中定义 多个 @DltHandler 方法，Spring 容器启动时会抛出 IllegalStateException（歧义绑定）。生产中通过以下模式避免冲突：

模式1：按业务隔离（不同类/容器工厂）

将不同业务的重试与死信处理逻辑拆分到 不同类，或使用 不同容器工厂 隔离：

1// 订单业务消费者（独立类）
2@Service
3public class OrderConsumer {
4    @RetryableTopic(/* 订单重试配置 */)
5    @KafkaListener(topics = "order-topic")
6    public void processOrder(Order order) { /* ... */ }
7
8    @DltHandler // 仅处理 OrderConsumer 的重试失败消息
9    public void handleOrderDlt(Order order) { /* ... */ }
10}
11
12// 支付业务消费者（独立类）
13@Service
14public class PaymentConsumer {
15    @RetryableTopic(/* 支付重试配置 */)
16    @KafkaListener(topics = "payment-topic")
17    public void processPayment(Payment payment) { /* ... */ }
18
19    @DltHandler // 仅处理 PaymentConsumer 的重试失败消息
20    public void handlePaymentDlt(Payment payment) { /* ... */ }
21}
22

模式2：显式指定死信 Topic（高级配置）

通过 DeadLetterPublishingRecoverer 自定义死信 Topic 名称，脱离默认拼接规则（适用于跨类绑定）：

1@Configuration
2public class KafkaErrorConfig {
3    @Bean
4    public ConcurrentKafkaListenerContainerFactory<String, Object> customContainerFactory(
5            ConsumerFactory<String, Object> consumerFactory,
6            KafkaTemplate<String, Object> kafkaTemplate) {
7        
8        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
9            new ConcurrentKafkaListenerContainerFactory<>();
10        factory.setConsumerFactory(consumerFactory);
11        
12        // 自定义死信发布器（指定死信 Topic 名称）
13        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
14            kafkaTemplate, 
15            (record, ex) -> new TopicPartition(record.topic() + "-custom-dlq", record.partition()) // 自定义 DLQ 名称
16        );
17        
18        // 配置错误处理器（重试+死信）
19        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(2000L, 3));
20        factory.setCommonErrorHandler(errorHandler);
21        return factory;
22    }
23}
24

二、死信队列 Topic 名称拼接规则

死信队列（DLQ）的 Topic 名称由 原始 Topic 名称 和 死信后缀 拼接而成，支持默认规则和自定义配置。

1. 默认拼接规则

默认后缀：-dlt（Spring Kafka 内置默认值）。
拼接公式：${原始Topic名称} + ${默认后缀}。

示例：

原始 Topic：order-topic → 默认 DLQ Topic：order-topic-dlt。
原始 Topic：user-login → 默认 DLQ Topic：user-login-dlt。

2. 自定义后缀（`@RetryableTopic` 配置）

通过 @RetryableTopic 的 dltTopicSuffix 属性自定义后缀，覆盖默认值：

1@RetryableTopic(
2    attempts = "3", 
3    dltTopicSuffix = "-dead-letter" // 自定义后缀为 "-dead-letter"
4)
5@KafkaListener(topics = "order-topic")
6public void processOrder(Order order) { /* ... */ }
7
8// 死信 Topic 名称：order-topic-dead-letter（原始Topic + 自定义后缀）
9

3. 完整 Topic 名称生成逻辑（源码级解析）

Spring Kafka 通过 DeadLetterTopicResolver 接口实现 Topic 名称解析，核心逻辑如下：

1// 伪代码：死信 Topic 名称生成
2String originalTopic = record.topic(); // 原始 Topic 名称
3String suffix = determineSuffix(annotation); // 从 @RetryableTopic 获取 dltTopicSuffix（默认 "-dlt"）
4String dlqTopic = originalTopic + suffix; // 拼接结果
5

4. 生产环境常用命名规范

为避免 Topic 名称混乱，生产实践中建议遵循以下规范：

场景	命名示例	说明
默认死信队列	order-topic-dlt	简单直观，适合单一业务线
按环境隔离	order-topic-prod-dlt	区分环境（prod/test/dev）
按优先级隔离	order-topic-high-dlt	区分消息优先级（high/low）
跨集群死信	clusterA.order-topic-dlt	跨 Kafka 集群时添加集群标识

三、生产常用模式：重试+死信队列最佳实践

1. 模式1：基础重试+默认死信（简单业务）

适用场景：非核心业务（如日志收集），允许简单重试和默认死信。
配置要点：

使用默认 @RetryableTopic 和 @DltHandler。
依赖默认死信 Topic 名称（原始Topic-dlt）。

代码示例：

1@Service
2public class LogConsumer {
3    private static final Logger log = LoggerFactory.getLogger(LogConsumer.class);
4
5    // 重试配置：3次尝试（1次原始+2次重试），间隔1s
6    @RetryableTopic(
7        attempts = "3", 
8        backoff = @Backoff(delay = 1000)
9        // 不指定 dltTopicSuffix，使用默认 "-dlt"
10    )
11    @KafkaListener(topics = "app-log-topic", groupId = "log-group")
12    public void processLog(String logMessage) {
13        if (logMessage.contains("ERROR")) {
14            throw new RuntimeException("日志处理失败"); // 触发重试
15        }
16        log.info("处理日志: {}", logMessage);
17    }
18
19    // 死信处理：默认绑定到 processLog 的重试失败消息
20    @DltHandler
21    public void handleLogDlt(String logMessage, 
22                           @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
23        log.error("死信日志: {}, 错误: {}", logMessage, error);
24        // 保存到文件系统或低成本存储
25    }
26}
27

死信 Topic 名称：app-log-topic-dlt（原始 Topic app-log-topic + 默认后缀 -dlt）。

2. 模式2：自定义重试+独立死信 Topic（核心业务）

适用场景：核心业务（如订单支付），需明确死信 Topic 名称并隔离存储。
配置要点：

自定义 dltTopicSuffix（如 -order-dlq）。
通过容器工厂隔离不同业务的重试策略。

代码示例：

1@Configuration
2public class OrderKafkaConfig {
3    // 订单消费者容器工厂（自定义重试+死信）
4    @Bean
5    public ConcurrentKafkaListenerContainerFactory<String, Order> orderContainerFactory(
6            ConsumerFactory<String, Order> consumerFactory,
7            KafkaTemplate<String, Order> kafkaTemplate) {
8        
9        ConcurrentKafkaListenerContainerFactory<String, Order> factory = 
10            new ConcurrentKafkaListenerContainerFactory<>();
11        factory.setConsumerFactory(consumerFactory);
12        
13        // 自定义死信发布器（指定死信 Topic 后缀为 "-order-dlq"）
14        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
15            kafkaTemplate,
16            (record, ex) -> new TopicPartition(record.topic() + "-order-dlq", record.partition())
17        );
18        
19        // 重试策略：4次尝试（1+3），指数退避（1s→2s→4s）
20        DefaultErrorHandler errorHandler = new DefaultErrorHandler(
21            recoverer, 
22            new ExponentialBackOff(1000L, 2) // 初始1s，乘数2
23        );
24        errorHandler.setRetryListeners((record, ex, deliveryAttempt) -> 
25            log.warn("订单重试: 次数={}, 消息={}", deliveryAttempt, record.value())
26        );
27        
28        factory.setCommonErrorHandler(errorHandler);
29        return factory;
30    }
31}
32
33@Service
34public class OrderConsumer {
35    @RetryableTopic(
36        attempts = "4", 
37        containerFactory = "orderContainerFactory" // 使用自定义容器工厂
38        // 死信 Topic 名称由容器工厂的 DeadLetterPublishingRecoverer 决定
39    )
40    @KafkaListener(topics = "order-topic", groupId = "order-group")
41    public void processOrder(Order order) {
42        if (order.getAmount().compareTo(BigDecimal.valueOf(10000)) > 0) {
43            throw new BusinessException("金额超限"); // 触发重试
44        }
45        // 处理订单...
46    }
47
48    // 死信处理：绑定到 processOrder 的重试失败消息
49    @DltHandler
50    public void handleOrderDlt(Order order, 
51                           @Header(KafkaHeaders.DLT_EXCEPTION_CAUSE) Throwable ex) {
52        log.error("订单死信: ID={}, 错误={}", order.getOrderId(), ex.getMessage());
53        // 保存到数据库并触发人工审核
54    }
55}
56

死信 Topic 名称：order-topic-order-dlq（原始 Topic order-topic + 容器工厂自定义的 -order-dlq 后缀）。

3. 模式3：多级重试+分级死信（复杂业务）

适用场景：需区分“可重试异常”和“不可重试异常”，并路由到不同死信 Topic。
配置要点：

通过 include/exclude 指定重试异常类型。
对不同异常类型使用不同 DeadLetterPublishingRecoverer。

代码示例：

1@Configuration
2public class MultiLevelDltConfig {
3    @Bean
4    public ConcurrentKafkaListenerContainerFactory<String, Object> multiLevelContainerFactory(
5            ConsumerFactory<String, Object> consumerFactory,
6            KafkaTemplate<String, Object> kafkaTemplate) {
7        
8        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
9            new ConcurrentKafkaListenerContainerFactory<>();
10        factory.setConsumerFactory(consumerFactory);
11        
12        // 异常分类处理器
13        Map<Class<? extends Throwable>, String> exceptionMap = new HashMap<>();
14        exceptionMap.put(TransientException.class, "-transient-dlq"); // 瞬时异常→临时死信
15        exceptionMap.put(PermanentException.class, "-permanent-dlq"); // 永久异常→永久死信
16        
17        // 自定义死信发布器（按异常类型路由）
18        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
19            kafkaTemplate,
20            (record, ex) -> {
21                String suffix = exceptionMap.getOrDefault(ex.getClass(), "-default-dlq");
22                return new TopicPartition(record.topic() + suffix, record.partition());
23            }
24        );
25        
26        DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(3000L, 2));
27        errorHandler.addRetryableExceptions(TransientException.class); // 仅重试瞬时异常
28        errorHandler.addNotRetryableExceptions(PermanentException.class); // 永久异常不重试
29        
30        factory.setCommonErrorHandler(errorHandler);
31        return factory;
32    }
33}
34

四、关键注意事项

死信 Topic 需提前创建：
Kafka 不会自动创建死信 Topic，需通过命令行或代码提前创建（指定分区数和副本数）：

1bin/kafka-topics.sh --create --topic order-topic-dlt --bootstrap-server localhost:9092 --partitions 3 --replication-factor 2

避免死信 Topic 无限增长：
配置死信 Topic 的 消息保留策略（如保留 7 天），通过 log.retention.hours 控制。
监控死信队列堆积：
通过 Prometheus + Grafana 监控死信 Topic 的消息数量，设置告警阈值（如堆积超过 1000 条）。
死信消息人工介入：
核心业务的死信消息需定期人工审核，避免数据丢失（如订单死信需补单）。

总结

对应关系：@DltHandler 与 @RetryableTopic 需 一对一绑定（同一类或显式隔离），避免多 @DltHandler 冲突。
Topic 拼接：默认死信 Topic 名称为 ${原始Topic}-dlt，可通过 @RetryableTopic(dltTopicSuffix) 或 DeadLetterPublishingRecoverer 自定义。
生产模式：根据业务重要性选择基础重试（默认死信）、自定义死信（独立 Topic）或多级死信（分级路由），核心是 隔离、可观测、可恢复。

通过合理配置，可实现“重试兜底+死信救急”的完整消息可靠性保障体系。

《Kafka：消费者重试与死信队列的对应模式分析》是转载文章，点击查看原文。

Kafka：消费者重试与死信队列的对应模式分析

一、@RetryableTopic 与 @DltHandler 的对应关系

1. 基础对应规则（默认模式）

2. 多方法冲突与解决方案

模式1：按业务隔离（不同类/容器工厂）

模式2：显式指定死信 Topic（高级配置）

二、死信队列 Topic 名称拼接规则

1. 默认拼接规则

2. 自定义后缀（@RetryableTopic 配置）

3. 完整 Topic 名称生成逻辑（源码级解析）

4. 生产环境常用命名规范

三、生产常用模式：重试+死信队列最佳实践

1. 模式1：基础重试+默认死信（简单业务）

2. 模式2：自定义重试+独立死信 Topic（核心业务）

3. 模式3：多级重试+分级死信（复杂业务）

四、关键注意事项

总结

一、`@RetryableTopic` 与 `@DltHandler` 的对应关系

2. 自定义后缀（`@RetryableTopic` 配置）