spring-kafka 2.9.12中RetryableTopic主/重试/DLT主题不同并发配置咨询

Spring-Kafka 2.9.12确实没有原生支持为主主题和重试/DLT主题配置差异化并发数，但可以通过以下几种变通方案实现需求：

方案1：通过RetryTopicConfigurationBuilder定制容器属性 #

在构建RetryTopicConfiguration时，利用容器定制器针对不同类型的主题（主、重试、DLT）单独设置并发数。这种方式能保留RetryableTopic的自动重试逻辑，同时实现并发差异化：

@Bean
public RetryTopicConfiguration retryTopicConfig(KafkaTemplate<Object, Object> kafkaTemplate) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            // 针对重试主题定制容器并发
            .configureRetryTopics(retryConfig -> 
                retryConfig.containerCustomizer(container -> container.setConcurrency(2)))
            // 针对DLT定制容器并发
            .configureDlt(dltConfig -> 
                dltConfig.containerCustomizer(container -> container.setConcurrency(1)))
            .create(kafkaTemplate);
}

// 主主题消费者，设置主主题并发数
@KafkaListener(topics = "main-topic", concurrency = "3")
public void handleMainTopic(String message) {
    // 业务逻辑处理
}

如果需要区分不同重试次数的主题，可以通过判断主题名称（默认重试主题命名规则为{main-topic}-retry-{attempt}，DLT为{main-topic}-dlt）来调整对应容器的并发。

方案2：手动创建重试/DLT消费者容器 #

放弃RetryableTopic的自动容器生成，手动为主主题、重试主题、DLT主题分别定义消费者容器，完全自主控制每个容器的并发数。这种方式灵活性最高，但需要自行处理重试逻辑的路由：

// 主主题消费者，并发设为3
@KafkaListener(topics = "main-topic", concurrency = "3")
public void handleMain(String message) {
    try {
        processMessage(message);
    } catch (RetryableException e) {
        // 发送到一级重试主题
        kafkaTemplate.send("main-topic-retry-1", message);
    } catch (FatalException e) {
        // 直接发送到DLT
        kafkaTemplate.send("main-topic-dlt", message);
    }
}

// 一级重试主题消费者，并发设为2
@KafkaListener(topics = "main-topic-retry-1", concurrency = "2")
public void handleRetry1(String message) {
    try {
        processMessage(message);
    } catch (Exception e) {
        // 发送到DLT或下一级重试
        kafkaTemplate.send("main-topic-dlt", message);
    }
}

// DLT消费者，并发设为1
@KafkaListener(topics = "main-topic-dlt", concurrency = "1")
public void handleDlt(String message) {
    // DLT消息处理逻辑（比如记录日志、触发告警）
    log.error("DLT received failed message: {}", message);
}

方案3：使用BeanPostProcessor动态调整容器并发 #

利用Spring的BeanPostProcessor扩展点，在RetryableTopic自动创建的消费者容器初始化前，修改其并发属性。这种方式无需改动原有RetryableTopic配置，透明调整容器参数：

@Component
public class KafkaContainerConcurrencyAdjuster implements BeanPostProcessor {

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        if (bean instanceof AbstractMessageListenerContainer) {
            AbstractMessageListenerContainer<?, ?> container = (AbstractMessageListenerContainer<?, ?>) bean;
            String[] topics = container.getContainerProperties().getTopics();
            if (topics == null || topics.length == 0) {
                return bean;
            }
            String topic = topics[0];
            if (topic.contains("-retry-")) {
                // 所有重试主题并发设为2
                container.setConcurrency(2);
            } else if (topic.endsWith("-dlt")) {
                // DLT主题并发设为1
                container.setConcurrency(1);
            } else if ("main-topic".equals(topic)) {
                // 主主题并发设为3
                container.setConcurrency(3);
            }
        }
        return bean;
    }
}

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