基于Spring Boot 2.0与spring-kafka 2.1.x实现Kafka死信队列(DLQ)的最优方案

兄弟，你这个需求找对路子了！在Spring Boot 2.0搭配spring-kafka 2.1.x的场景下，实现死信队列（DLQ）且保证消息不丢失的最优方案，就是用官方原生支持的SeekToCurrentErrorHandler + DeadLetterPublishingRecoverer组合，这可是经过生产环境验证的标准玩法，完全能满足你「消息要么处理成功、要么进DLQ、要么重试直到成功」的要求。下面给你拆解具体实现步骤：

一、核心组件先搞懂 #

先理清楚两个关键组件的作用，避免瞎配置：

• DeadLetterPublishingRecoverer：专门负责把处理失败的消息转发到指定DLQ主题，还支持自定义DLQ命名规则（比如给原主题加后缀）
• SeekToCurrentErrorHandler：消费出现异常时，会让消费者重新定位到当前消息的偏移量（也就是重试），当重试次数耗尽后，就把消息交给DeadLetterPublishingRecoverer发送到DLQ

二、一步步配置实现 #

1. 依赖确认（Maven示例） #

确保spring-kafka版本和Spring Boot 2.0兼容，2.1.x系列完美适配，比如：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.1.14.RELEASE</version> <!-- 匹配Spring Boot 2.0.x的稳定版本 -->
</dependency>

2. 配置KafkaTemplate用于DLQ消息发送 #

DeadLetterPublishingRecoverer需要借助KafkaTemplate发送死信，先配置好这个Bean：

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

3. 核心ErrorHandler配置 #

这一步是关键，把两个组件串起来，设置重试次数和DLQ规则：

@Bean
public ErrorHandler kafkaErrorHandler(KafkaTemplate<String, Object> kafkaTemplate) {
    // 自定义DLQ主题规则：原主题名 + "-dlq"，比如原主题是"order-topic"，DLQ就是"order-topic-dlq"
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition(record.topic() + "-dlq", record.partition()));

    // 设置重试策略：间隔1秒重试，最多重试3次，之后发送到DLQ
    SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(1000L, 3L));

    // 重中之重：如果发送DLQ失败（比如网络故障、DLQ主题不存在），不提交原消息的偏移量
    // 这样消息会被重新消费，直到DLQ发送成功，彻底避免消息丢失
    errorHandler.setCommitRecovered(false);
    return errorHandler;
}

4. 让消费者使用这个ErrorHandler #

有两种配置方式：

方式一：单个消费者指定

在@KafkaListener注解里直接绑定errorHandler：

@KafkaListener(topics = "your-business-topic", groupId = "your-consumer-group", errorHandler = "kafkaErrorHandler")
public void consumeMessage(String message) {
    // 这里写业务处理逻辑，一旦抛出异常就会触发重试和DLQ流程
    if (someBusinessConditionFails) {
        throw new RuntimeException("业务处理失败，触发DLQ");
    }
}

方式二：全局配置所有消费者

如果要让所有@KafkaListener都用这个ErrorHandler，配置全局容器工厂即可：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> consumerFactory, ErrorHandler kafkaErrorHandler) {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setErrorHandler(kafkaErrorHandler); // 全局绑定ErrorHandler
    return factory;
}

三、关键细节保证不丢消息 #

• 重试策略灵活调整：上面用的FixedBackOff是固定间隔重试，怕短时间重试压垮系统的话，可以换成ExponentialBackOff实现指数退避（比如第一次等1秒，第二次2秒，第三次4秒）
• DLQ发送失败的兜底：setCommitRecovered(false)这个配置一定要加！如果发送DLQ时遇到网络问题或者DLQ主题不可用，原消息的偏移量不会被提交，消费者会重新拉取这条消息重试，直到DLQ恢复可用，绝对不会丢消息
• DLQ主题提前准备：生产环境别依赖Kafka自动创建主题，提前手动创建DLQ主题，配置合适的分区数和副本数，保证DLQ本身的高可用
• 偏移量提交逻辑：默认情况下，spring-kafka只有在消息处理成功（或者成功发送到DLQ且setCommitRecovered(true)）时才会提交偏移量，失败的消息不会提交，确保消息不会被跳过

四、测试验证场景 #

建议你测试以下几个场景，确保符合预期：

• 处理成功：消息正常消费，偏移量正常提交，不会进入DLQ
• 处理失败重试后进入DLQ：业务抛出异常，重试3次后消息出现在DLQ主题，原主题偏移量提交
• DLQ发送失败：故意让DLQ主题不可用，此时消息会被反复重试，直到DLQ恢复，消息成功发送后才会提交偏移量

内容的提问来源于stack exchange，提问作者Eugene Khyst