如何在WebFlux Reactive Kafka中实现死信队列功能

现有代码存在的核心问题 #

• 流级别的重试配置逻辑错误：原代码外层的retryWhen(Retry.backoff(30, Duration.of(10, ChronoUnit.SECONDS)))是对整个消费流生效，单条消息消费失败会触发整个流重新订阅，导致其他正常消息被重复消费，坏消息会持续阻塞整个消费链路
• 异常处理逻辑缺失：原代码的onErrorContinue仅打印错误日志，没有重试耗尽后的消息转发逻辑
• 重试实现存在失效风险：两个consumeWithRetry版本中，不带Mono.defer的版本1无法处理consume方法抛出的同步异常，重试操作符不会触发重新执行，必须采用版本2的defer写法
• 元数据丢失：原代码map阶段只取了消息value，丢失了原始消息的key、分区、偏移量、主题等元数据，不利于死信排查和后续重放

具体实现步骤 #

1. 基础依赖准备 #

提前配置好用于发送死信的ReactiveKafkaProducerTemplate，序列化规则和业务消费的消息序列化规则保持一致，提前创建好对应业务的死信主题，建议命名规则为原业务主题名.DLQ。

2. 重构消费主链路 #

移除流级别的全局重试配置，将单条消息的异常处理、重试逻辑全部收敛在单条消息的flatMap作用域内，避免单消息异常影响整个消费流：

private static final String DLQ_TOPIC = "your-business-topic.DLQ";
private static final int MAX_RETRY_COUNT = 2;

@Autowired
private ReactiveKafkaProducerTemplate<String, MessageRecord> dlqProducerTemplate;

// 消费主逻辑
reactiveKafkaConsumerTemplate
        .receiveAutoAck()
        .flatMap(consumerRecord -> {
            MessageRecord messageValue = consumerRecord.value();
            return consumeWithRetry(messageValue)
                    // 重试耗尽后触发DLQ转发，不把异常抛到外层流
                    .onErrorResume(ex -> forwardToDlq(consumerRecord, ex));
        })
        // 仅兜底处理流级别的未预期异常，不处理单消息消费失败
        .onErrorContinue((ex, obj) -> log.error("消费流触发未预期异常, 异常信息:{}", ex.getMessage(), ex))
        .subscribe();

3. 修正单消息重试逻辑 #

采用带Mono.defer的重试实现，保证同步异常、异步异常都能被重试逻辑正确捕获触发重试：

public Mono<Void> consumeWithRetry(MessageRecord message) {
    return Mono.defer(() -> consume(message))
            .retry(MAX_RETRY_COUNT);
}

如果需要指数退避、重试间隔等更灵活的重试策略，可以替换retry(2)为自定义的retryWhen配置，只要保证重试次数耗尽后异常能正常抛出触发后续DLQ转发即可。

4. 实现死信转发逻辑 #

转发死信时建议保留原始消息的key，保证和原消息分区路由规则一致，同时打印完整的异常栈、原始消息偏移量、分区信息方便排查：

private Mono<Void> forwardToDlq(ConsumerRecord<String, MessageRecord> originRecord, Throwable consumeEx) {
    String messageKey = originRecord.key();
    MessageRecord failedMessage = originRecord.value();
    log.error("消息达到最大重试次数{}, 转发至死信主题, 偏移量:{}, 分区:{}, 异常:{}",
            MAX_RETRY_COUNT,
            originRecord.offset(),
            originRecord.partition(),
            consumeEx.getMessage(),
            consumeEx);
    
    return dlqProducerTemplate.send(DLQ_TOPIC, messageKey, failedMessage)
            .doOnError(sendEx -> log.error("死信发送失败, 消息偏移量:{}", originRecord.offset(), sendEx))
            .then();
}

关键注意事项 #

• 必须删除原代码中外层的全局retryWhen配置：全局重试会导致单条坏消息触发整个消费流反复重连，所有正常消息都会被重复消费，完全破坏消费链路的稳定性
• 禁止在flatMap外层处理单消息消费异常：单条消息的消费、重试、DLQ转发逻辑必须完全在flatMap内部通过onErrorResume兜底，一旦异常流出flatMap就会终止整个流的消费
• 死信消息不要随意修改原始value和key：如果需要附加异常信息、重试次数等元数据，建议通过Kafka消息头传递，避免修改原始消息内容导致后续重放时反序列化失败
• 如果需要保证死信发送的可靠性，可以配置生产者的acks=all以及合适的重试参数，避免死信本身发送失败导致消息丢失

内容的提问来源于stack exchange，提问作者Rambabu Kokkiligadda