Spring Kafka 2.9.6：如何解耦消费处理并复用RetryableTopic重试逻辑？

完全可以实现你的需求，且DLT（死信主题）会和正常RetryableTopic流程一样自动生效。以下是具体实现要点：

一、复用RetryableTopic的重试逻辑 #

Spring Kafka的RetryableTopic特性核心依赖RetryTopicSender和DestinationTopicResolver这两个组件：前者负责将消息发送至重试主题并自动添加重试元数据头（如重试次数kafka_retry_topic_attempt），后者负责解析重试主题、DLT的路由规则。你可以直接注入这两个组件，在业务处理异常时手动触发重试逻辑。

关键代码示例 #

1. 启用RetryableTopic并配置重试/DLT规则：

@Configuration
@EnableRetryTopic
public class KafkaRetryConfig {
    @Bean
    public RetryTopicConfiguration retryTopicConfig(KafkaTemplate<String, Object> kafkaTemplate) {
        return RetryTopicConfigurationBuilder
                .newInstance()
                .fixedBackOff(5000) // 固定退避5秒
                .maxAttempts(3) // 最大重试3次
                .dltHandlerMethod("dltConsumer", "handleDlt") // 指定DLT处理方法
                .create(kafkaTemplate);
    }
}

2. 在业务处理器中注入RetryTopicSender，异常时触发重试：

@Component
public class MessageProcessor {
    private final RetryTopicSender retryTopicSender;
    private final ExecutorService processingThreadPool = Executors.newFixedThreadPool(10);

    public MessageProcessor(RetryTopicSender retryTopicSender) {
        this.retryTopicSender = retryTopicSender;
    }

    // 接收消费端传递的消息，委托线程池处理
    public void delegateProcessing(ConsumerRecord<String, Object> record) {
        processingThreadPool.submit(() -> {
            try {
                // 执行业务处理逻辑
                executeBusinessLogic(record);
            } catch (Exception e) {
                // 手动触发重试：自动添加重试头并发送至对应重试主题
                retryTopicSender.sendToRetryTopic(record, e);
            }
        });
    }

    private void executeBusinessLogic(ConsumerRecord<String, Object> record) {
        // 你的业务处理代码，模拟失败场景
        throw new RuntimeException("业务处理失败，触发重试");
    }
}

二、解耦消费与处理的实现 #

按照你“拉取单条消息→委托线程池→提交偏移量→重复流程”的需求，需注意以下配置：

• 单条消息拉取：在消费者配置中设置max.poll.records=1，确保每次poll仅获取一条消息。
• 偏移量提交时机：拉取消息后立即提交偏移量（同步/异步均可），再将消息委托至线程池处理。此方式将消费确认与业务处理解耦，即使业务处理失败，原主题的偏移量已提交，不会重复消费，而是通过重试主题重新触发处理。

消费端拉取逻辑示例 #

@Component
public class MessageConsumer {
    private final Consumer<String, Object> kafkaConsumer;
    private final MessageProcessor messageProcessor;

    public MessageConsumer(@Qualifier("kafkaConsumerFactory") ConsumerFactory<String, Object> consumerFactory,
                           MessageProcessor messageProcessor) {
        this.kafkaConsumer = consumerFactory.createConsumer();
        this.messageProcessor = messageProcessor;
        // 订阅原始主题
        this.kafkaConsumer.subscribe(Collections.singletonList("original-topic"));
    }

    // 定时拉取消息，间隔可根据业务调整
    @Scheduled(fixedDelay = 100)
    public void pollMessage() {
        ConsumerRecords<String, Object> records = kafkaConsumer.poll(Duration.ofMillis(100));
        if (!records.isEmpty()) {
            ConsumerRecord<String, Object> record = records.iterator().next();
            // 提交偏移量，确认已拉取该消息
            kafkaConsumer.commitSync(Collections.singletonMap(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
            ));
            // 委托线程池处理
            messageProcessor.delegateProcessing(record);
        }
    }
}

三、DLT的自动工作验证 #

只要你在RetryTopicConfiguration中正确配置了DLT规则，当重试次数耗尽后，RetryTopicSender会自动将消息转发至DLT，完全复用正常RetryableTopic的DLT路由逻辑，无需额外开发。

DLT消费者示例 #

@Component
public class DltConsumer {
    public void handleDlt(ConsumerRecord<String, Object> record) {
        // 处理死信消息的逻辑，如记录日志、告警等
        System.out.println("处理死信消息: " + record.value() + ", 重试次数耗尽");
    }
}

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