Spring Cloud Stream Kafka Binder多DeadLetterPublishingRecoverer配置咨询

首先，你之前的全局CustomBatchErrorRecovererHandler不生效的核心原因有两点：

1. Spring Cloud Stream的每个消费者绑定可以独立配置错误处理器，全局@Component标记的错误处理器会被所有绑定共用，且你通过消息key判断绑定的逻辑不可靠（批量消费的消息都来自当前绑定对应的主题，无需通过key区分）；
2. 没有为每个绑定明确指定专属的错误处理器，导致Spring默认复用了同一个死信恢复器Bean。

正确的实现步骤： #

1. 定义命名的DeadLetterPublishingRecoverer Bean

为两个消费者分别创建独立的死信恢复器，通过@Bean的名称区分：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.support.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DlqConfig {

    // 对应第一个消费者绑定的死信恢复器，发送到topic1的死信主题
    @Bean("dlqRecovererForTopic1")
    public DeadLetterPublishingRecoverer dlqRecovererForTopic1(KafkaTemplate<?, ?> kafkaTemplate) {
        return new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition("topic1-dlq", record.partition()));
    }

    // 对应第二个消费者绑定的死信恢复器，发送到topic2的死信主题
    @Bean("dlqRecovererForTopic2")
    public DeadLetterPublishingRecoverer dlqRecovererForTopic2(KafkaTemplate<?, ?> kafkaTemplate) {
        return new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition("topic2-dlq", record.partition()));
    }
}

2. 为每个绑定创建专属的BatchErrorHandler

基于各自的死信恢复器，创建独立的批量错误处理器，同样用Bean名称区分：

import org.springframework.kafka.listener.BatchErrorHandler;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.factory.annotation.Qualifier;

@Configuration
public class BatchErrorHandlerConfig {

    // 第一个绑定的批量错误处理器
    @Bean("batchErrorHandlerForTopic1")
    public BatchErrorHandler batchErrorHandlerForTopic1(
            @Qualifier("dlqRecovererForTopic1") DeadLetterPublishingRecoverer recoverer) {
        return (exception, records) -> records.forEach(record -> recoverer.accept(record, exception));
    }

    // 第二个绑定的批量错误处理器
    @Bean("batchErrorHandlerForTopic2")
    public BatchErrorHandler batchErrorHandlerForTopic2(
            @Qualifier("dlqRecovererForTopic2") DeadLetterPublishingRecoverer recoverer) {
        return (exception, records) -> records.forEach(record -> recoverer.accept(record, exception));
    }
}

3. 为每个消费者绑定指定对应的错误处理器

通过配置文件（以YAML为例），为每个绑定明确指定专属的错误处理器Bean名称：

spring:
  cloud:
    stream:
      bindings:
        # 第一个消费者绑定配置
        consumer-topic1-in:
          destination: topic1
          group: group-topic1
          consumer:
            batch-mode: true
            max-batch-size: 100
        # 第二个消费者绑定配置
        consumer-topic2-in:
          destination: topic2
          group: group-topic2
          consumer:
            batch-mode: true
            max-batch-size: 100
      kafka:
        bindings:
          consumer-topic1-in:
            consumer:
              batch-error-handler: batchErrorHandlerForTopic1
          consumer-topic2-in:
            consumer:
              batch-error-handler: batchErrorHandlerForTopic2

替代方案：编程式配置错误处理器 #

如果你偏好代码配置，可以通过ConsumerCustomizer为每个绑定动态设置错误处理器：

import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
import org.springframework.cloud.stream.binder.ConsumerCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.listener.BatchErrorHandler;

@Configuration
public class StreamCustomizerConfig {

    @Bean
    public ConsumerCustomizer<KafkaMessageChannelBinder> consumerCustomizer(
            @Qualifier("batchErrorHandlerForTopic1") BatchErrorHandler handler1,
            @Qualifier("batchErrorHandlerForTopic2") BatchErrorHandler handler2) {
        return (bindingName, consumerFactory) -> {
            // 根据绑定名称匹配对应的错误处理器
            if ("consumer-topic1-in".equals(bindingName)) {
                consumerFactory.getContainerProperties().setBatchErrorHandler(handler1);
            } else if ("consumer-topic2-in".equals(bindingName)) {
                consumerFactory.getContainerProperties().setBatchErrorHandler(handler2);
            }
        };
    }
}

这样配置后，每个消费者绑定出错时，会调用各自对应的死信恢复器，将消息发送到专属的死信主题，不会再出现复用同一个恢复器的问题。

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