如何为Spring Kafka的@RetryableTopic指定自定义死信队列（DLT）

要覆盖@RetryableTopic默认的DLT命名规则（主主题+"_dlt"），直接用配置文件的spring.kafka.consumer.template.dead-letter-topic是不生效的——因为这个配置是给DeadLetterPublishingRecoverer单独使用的，而@RetryableTopic有一套独立的主题命名机制。你可以通过以下两种方式实现自定义DLT名称：

方法一：直接指定DLT名称（最简方案） #

通过RetryTopicConfigurationBuilder直接配置自定义DLT名称，无需额外实现类：

1. 创建Kafka重试配置类：

@Configuration
public class KafkaRetryConfig {

    @Bean
    public RetryTopicConfiguration retryTopicConfiguration(ConsumerFactory<?, ?> consumerFactory,
                                                           @Value("${spring.kafka.consumer.template.dead-letter-topic}") String customDltTopic) {
        return RetryTopicConfigurationBuilder
                .newInstance()
                // 直接指定自定义DLT名称
                .dltTopicName(customDltTopic)
                // 对齐你原@RetryableTopic的配置
                .fixedDelayTopicStrategy(FixedDelayStrategy.SINGLE_TOPIC)
                .maxAttempts(4)
                .backoff(Backoff.of(Duration.ofMillis(1000)))
                .autoCreateTopics(false)
                .suffixTopicsWithDelayValue()
                .create(consumerFactory);
    }
}

2. 修改监听器注解：
   去掉原方法上的@RetryableTopic，保留@KafkaListener即可：

@KafkaListener(topics = "${spring.kafka.template.default-topic}")
public void processMessage(String message) {
    // 业务逻辑处理
    throw new RuntimeException("模拟消息处理失败，触发重试");
}

方法二：自定义主题命名策略（灵活扩展） #

如果需要同时自定义重试主题和DLT的命名规则，实现TopicNamingStrategy接口：

1. 实现自定义命名策略：

@Component
public class CustomTopicNamingStrategy implements TopicNamingStrategy {

    @Value("${spring.kafka.consumer.template.dead-letter-topic}")
    private String customDltTopic;

    @Override
    public String getRetryTopicName(String originalTopic, int attempt, long delay) {
        // 保留原重试主题的命名逻辑（主主题+延迟值后缀）
        return originalTopic + "-" + delay;
    }

    @Override
    public String getDltTopicName(String originalTopic) {
        // 返回自定义DLT名称
        return customDltTopic;
    }
}

2. 在配置类中关联自定义策略：

@Configuration
public class KafkaRetryConfig {

    @Bean
    public RetryTopicConfiguration retryTopicConfiguration(ConsumerFactory<?, ?> consumerFactory,
                                                           CustomTopicNamingStrategy customNamingStrategy) {
        return RetryTopicConfigurationBuilder
                .newInstance()
                .customTopicNamingStrategy(customNamingStrategy)
                .fixedDelayTopicStrategy(FixedDelayStrategy.SINGLE_TOPIC)
                .maxAttempts(4)
                .backoff(Backoff.of(Duration.ofMillis(1000)))
                .autoCreateTopics(false)
                .create(consumerFactory);
    }
}

3. 同样去掉监听器上的@RetryableTopic注解，保留@KafkaListener。

注意事项 #

• 确保使用的Spring Kafka版本在2.8及以上，以上API是该版本后引入的。
• 如果你仍想保留@RetryableTopic注解，也可以通过@EnableRetryTopics配合RetryTopicConfigurer来注册自定义配置，但用RetryTopicConfigurationBuilder的方式更直观。

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