Spring Cloud Stream如何为多源Topic配置对应DLQ？

你遇到的问题根源在于：单个输入绑定无法同时为多个Topic配置独立的DLQ，且Kafka Topic名称不允许包含逗号，这也是触发InvalidTopicException的直接原因。要实现每个源Topic对应专属DLQ，需为每个Topic创建独立的输入绑定，再让同一个function处理这些绑定的消息。

具体配置步骤 #

1. 拆分输入绑定并配置独立DLQ #

将原单个functionA-in-0拆分为两个输入绑定，分别对应topic-1和topic-2，并为每个绑定单独配置DLQ参数：

spring:
  cloud:
    stream:
      bindings:
        # 对应topic-1的输入绑定
        functionA-in-0:
          destination: topic-1
          binder: kafka
          group: local.kafka-sink
          consumer:
            max-attempts: 3
        # 对应topic-2的输入绑定
        functionA-in-1:
          destination: topic-2
          binder: kafka
          group: local.kafka-sink
          consumer:
            max-attempts: 3
      kafka:
        binder:
          brokers: localhost:9092
        bindings:
          # 为topic-1的绑定配置专属DLQ
          functionA-in-0:
            consumer:
              enable-dlq: true
              dlq-name: topic-1-dlq
          # 为topic-2的绑定配置专属DLQ
          functionA-in-1:
            consumer:
              enable-dlq: true
              dlq-name: topic-2-dlq

2. 保持Function兼容多输入 #

Spring Cloud Stream默认支持函数接收多个输入，只要绑定名称遵循{functionName}-in-{index}的命名规则（index从0开始递增）。你的functionA无需额外修改，示例如下：

import java.util.function.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class StreamFunctionConfig {

    @Bean
    public Consumer<String> functionA() {
        return message -> {
            // 自定义消息处理逻辑
            System.out.println("Received message: " + message);
            // 模拟异常触发DLQ（按需保留）
            // if (someErrorCondition) throw new RuntimeException("Processing failed");
        };
    }
}

配置说明 #

• 每个输入绑定对应一个源Topic，独立配置DLQ后，各自的错误消息会被精准路由到对应DLQ（topic-1-dlq和topic-2-dlq）。
• 两个绑定共用同一个消费组local.kafka-sink，既保证消费进度独立管理，又符合业务分组需求。
• 彻底规避了单个绑定配置多Topic、多DLQ的冲突，同时解决了Kafka Topic名称含逗号的非法问题。

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