Spring Kafka非阻塞重试：发送至重试主题失败的处理方案咨询

当使用Spring Kafka的非阻塞重试机制时，若发送重试主题因连接问题等原因失败，除了调整Producer的retries配置外，还有以下几种更具针对性的方案：

1. 自定义ProducerListener捕获发送失败并兜底 #

Spring Kafka的KafkaTemplate允许绑定自定义的ProducerListener，用来监听消息发送的成功/失败事件。当发送重试主题失败时，你可以在这里记录详细日志、触发告警，甚至将消息暂存到本地存储（数据库、文件）做后续补偿。

示例代码：

@Component
public class RetryTopicFailureListener implements ProducerListener<String, Object> {
    private static final Logger log = LoggerFactory.getLogger(RetryTopicFailureListener.class);

    @Override
    public void onError(ProducerRecord<String, Object> record, Exception exception) {
        log.error("发送重试主题失败，消息Key: {}, 内容: {}", record.key(), record.value(), exception);
        // 这里添加兜底逻辑：比如写入补偿表、发送告警通知
    }
}

然后给KafkaTemplate绑定这个监听器：

@Bean
public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory,
                                                   RetryTopicFailureListener failureListener) {
    KafkaTemplate<String, Object> template = new KafkaTemplate<>(producerFactory);
    template.setProducerListener(failureListener);
    return template;
}

2. 本地有限重试+死信队列兜底 #

如果发送重试主题的失败是临时的，可以用Spring Retry在本地做几次重试，若仍失败则直接转发到死信队列（DLQ），避免消息卡在重试环节。

示例代码（自定义Recoverer）：

@Bean
public DeadLetterPublishingRecoverer deadLetterRecoverer(KafkaTemplate<String, Object> kafkaTemplate) {
    // 基础的重试主题转发逻辑
    DeadLetterPublishingRecoverer baseRecoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition("retry-topic", record.partition()));

    // 配置本地重试模板（最多3次）
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));

    // 包装成带本地重试的Recoverer
    return (record, ex) -> retryTemplate.execute(context -> {
        try {
            baseRecoverer.accept(record, ex);
            return null;
        } catch (Exception retryEx) {
            // 本地重试失败，转发到死信队列
            kafkaTemplate.send("dlq-topic", record.key(), record.value());
            throw retryEx;
        }
    });
}

3. 开启Producer幂等性+事务保障 #

如果业务对消息投递可靠性要求高，可以开启Producer的幂等性和事务，确保消息至少一次投递。当发送重试主题失败时，事务会自动回滚，避免消息丢失，后续可以通过事务日志或补偿机制重试。

配置示例（application.properties）：

# 开启幂等性
spring.kafka.producer.enable-idempotence=true
# 设置事务ID前缀
spring.kafka.producer.transaction-id-prefix=kafka-tx-

然后在重试逻辑中使用事务：

@Autowired
private KafkaTemplate<String, Object> kafkaTemplate;

@Transactional
public void forwardToRetryTopic(ConsumerRecord<String, Object> record) {
    try {
        kafkaTemplate.send("retry-topic", record.key(), record.value());
    } catch (Exception e) {
        // 发送失败会触发事务回滚，消息可后续重新处理
        throw new RuntimeException("转发重试主题失败", e);
    }
}

4. 持久化补偿+定时重试 #

如果Kafka集群长时间不可用，最好将失败的消息持久化到数据库或分布式存储（如Redis），然后通过定时任务定期扫描并重试发送，确保消息不会丢失。

核心思路：

• 创建一张补偿表，字段包含：消息ID、原主题、重试主题、消息内容、重试次数、状态、创建时间
• 发送重试主题失败时，将消息插入补偿表
• 用@Scheduled注解编写定时任务，每隔一段时间扫描状态为"失败"的消息，调用KafkaTemplate重新发送；成功后更新状态，超过最大重试次数则标记为死信

总结 #

建议将多种方案组合使用：先用Producer的retries配置处理轻微的临时连接问题，再通过自定义ProducerListener捕获剩余失败，结合本地重试和持久化补偿，最后用死信队列兜底，形成一套完整的容错机制，既保证消息不丢失，又避免无效重试占用资源。

内容的提问来源于stack exchange，提问作者Oleg K.