能否将ReactiveKafkaConsumerTemplate用于DeadLetterPublishingRecoverer？

首先明确两个核心点：

• DeadLetterPublishingRecoverer是Spring Kafka同步版的组件，它依赖的KafkaOperations接口由同步的KafkaTemplate实现，作用是完成死信消息的同步发送。
• ReactiveKafkaConsumerTemplate是Reactive Kafka的消费端模板，只负责消费消息，不具备生产能力，所以绝对不能直接传入DeadLetterPublishingRecoverer的构造参数中。

如果你整条链路都基于Reactive Kafka，推荐采用Reactive生态原生的死信处理方案：

• 在Reactive消费流中通过onErrorResume或retryWhen捕获异常，再用ReactiveKafkaProducerTemplate将死信发送到指定topic，示例代码如下：

@Autowired
private ReactiveKafkaProducerTemplate<String, Object> reactiveProducerTemplate;

public Flux<ConsumerRecord<String, Object>> consume() {
    return reactiveConsumerTemplate.receiveAutoAck()
            .flatMap(record -> processRecord(record)
                    .onErrorResume(ex -> {
                        // 构造死信消息，可添加错误相关头信息
                        ProducerRecord<String, Object> dlqRecord = new ProducerRecord<>("dlq-topic", record.key(), record.value());
                        dlqRecord.headers().add("error-detail", ex.getMessage().getBytes(StandardCharsets.UTF_8));
                        // 异步发送死信
                        return reactiveProducerTemplate.send(dlqRecord)
                                .doOnSuccess(sendResult -> log.info("死信发送成功，topic: {}", sendResult.recordMetadata().topic()))
                                .then(Mono.empty());
                    })
            );
}

如果非要在同步的ConcurrentKafkaListenerContainerFactory中配合Reactive模板使用，有个不推荐的折衷方案：

• 自行编写适配类，将ReactiveKafkaProducerTemplate包装成KafkaOperations接口的实现，但这个适配类里的send方法需要阻塞等待响应，会破坏Reactive的非阻塞特性，违背Reactive设计初衷。

总结：既然已经采用Reactive Kafka，建议遵循Reactive生态的错误处理流程，不要强行套用同步版的DeadLetterPublishingRecoverer，既能契合生态设计，也能保留非阻塞的优势。

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