Spring Kafka实现3次消息重试及本地死信队列功能咨询

你当前的代码已经完成了基础配置，只需针对「发送端失败捕获」、「死信队列存储」两个核心点做少量修改即可满足需求，修改步骤如下：

1. 先实现本地死信队列存储类 #

先定义一个线程安全的死信队列管理组件，当前给出内存级实现，如需持久化可自行替换为数据库、本地文件存储逻辑：

import org.springframework.stereotype.Component;
import java.util.concurrent.ConcurrentLinkedQueue;

@Component
public class LocalDeadLetterQueue {
    // 线程安全队列存储死信消息
    private final ConcurrentLinkedQueue<String> dlqStorage = new ConcurrentLinkedQueue<>();

    // 存入失败消息
    public void saveFailedMessage(String message) {
        dlqStorage.offer(message);
        // 如需持久化，可在此处添加写入本地文件/数据库的逻辑
    }

    // 后续可扩展获取死信消息、重发等能力
    public ConcurrentLinkedQueue<String> getAllFailedMessages() {
        return dlqStorage;
    }
}

2. 改造发送端，捕获发送失败的消息 #

你当前配置的ProducerConfig.RETRIES_CONFIG = 3已经实现了Kafka客户端内部自动重试3次的能力，只需添加发送结果回调，捕获重试后仍然失败的消息存入死信队列即可：
修改KafkaApplication的TestBean类：

public static class TestBean {
    @Autowired
    private KafkaTemplate<String, String> template;
    @Autowired
    private LocalDeadLetterQueue deadLetterQueue;

    public void send(String message) {
        this.template.send(AppConfiguration.topic, message).addCallback(
            // 发送成功回调
            successResult -> System.out.println("消息发送成功：" + message),
            // 发送失败回调（已经过Kafka内部3次重试）
            failResult -> {
                System.out.println("消息发送重试耗尽，存入本地死信队列：" + message);
                deadLetterQueue.saveFailedMessage(message);
            }
        );
        this.template.flush();
    }
}

注意：ProducerConfig.RETRIES_CONFIG配置的是重试次数，加上首次发送，总共会尝试4次，如果你需要总尝试次数为3次，把该配置值改为2即可。

3. 修正消费端重试逻辑（可选） #

你当前消费者代码中catch了所有异常，导致Spring配置的重试模板无法触发（异常没有抛到容器层），如果需要消费失败也触发3次重试、失败后存入死信队列，做两处修改：

3.1 修改消费者监听方法，不要吞异常 #

@KafkaListener(topics = AppConfiguration.topic, groupId = AppConfiguration.groupid, containerFactory = "kafkaListenerContainerFactory")
public void listen(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
    if (message.startsWith("fail")) {
        throw new RuntimeException("failed");
    }
    System.out.println("Successfully Received: " + message + " (partition: " + partition + ")");
    this.latch.countDown();
}

3.2 修改Kafka配置类的重试恢复回调，存入死信队列 #

先注入死信队列组件，再修改kafkaListenerContainerFactory的恢复逻辑：

@Autowired
private LocalDeadLetterQueue deadLetterQueue;

@Bean
ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory, RetryTemplate retryTemplate) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, (ConsumerFactory<Object, Object>) kafkaConsumerFactory);
    factory.setRetryTemplate(retryTemplate);
    factory.setRecoveryCallback(context -> {
        // 从上下文获取失败的消费记录
        ConsumerRecord<?, ?> failedRecord = (ConsumerRecord<?, ?>) context.getAttribute("record");
        String failedMsg = (String) failedRecord.value();
        log.error("消费重试次数耗尽，消息存入死信队列：{}", failedMsg);
        deadLetterQueue.saveFailedMessage(failedMsg);
        return null;
    });
    return factory;
}

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