Spring Cloud Stream同Payload双Kafka流映射异常原因及解决方法

我尝试通过Spring Cloud Stream将两个包含相同消息Payload类型的独立Kafka流，映射到两个均使用同一服务实现类的不同Consumer Bean。代码与配置如下：

代码示例 #

@Bean
public Consumer<MyPayload> myPayloadConsumerFromTopic1(MyPayloadProcessor myPayloadProcessor) {
  return myPayloadProcessor::processMyPayload;
}

@Bean
public Consumer<MyPayload> myPayloadConsumerFromTopic2(MyPayloadProcessor myPayloadProcessor) {
  return myPayloadProcessor::processMyPayload;
}

配置示例 #

spring:
  cloud:
    function:
      definition: myPayloadConsumerFromTopic1,myPayloadConsumerFromTopic2
    stream:
      bindings:
        myPayloadConsumerFromTopic1-in-0: 
          destination: topic-1
          group: myPayloadConsumerFromTopic1Group
          ...
        myPayloadConsumerFromTopic2-in-0: 
          destination: topic-2
          group: myPayloadConsumerFromTopic2Group
          ...

启动微服务时出现错误：

org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'my-payload-processor-1.myPayloadConsumerFromTopic1myPayloadConsumerFromTopic2-in-0'

这个错误的核心原因是Spring Cloud Stream默认会对同类型的函数进行聚合：你定义的两个Consumer<MyPayload>函数属于同一类型，框架会自动将它们合并成一个复合函数，导致通道名称被错误拼接为myPayloadConsumerFromTopic1myPayloadConsumerFromTopic2-in-0。而你的配置中只给两个独立的函数通道做了绑定，合并后的通道没有对应的订阅者，因此抛出"Dispatcher has no subscribers"异常。

这种配置方式本身是被支持的，但需要调整配置避免函数被自动聚合。

方案1：禁用函数聚合 #

在配置中添加以下参数，关闭Spring Cloud Stream的函数聚合功能，让每个函数独立处理：

spring:
  cloud:
    stream:
      function:
        aggregation:
          enabled: false

添加后，框架会分别处理两个Consumer函数，你的原有绑定配置会生效，两个通道会分别订阅对应的Kafka Topic。

方案2：显式指定函数绑定（可选） #

如果需要保留聚合功能，也可以通过显式配置每个函数的输入通道来避免名称拼接问题：

spring:
  cloud:
    stream:
      function:
        bindings:
          myPayloadConsumerFromTopic1-in-0: myPayloadConsumerFromTopic1-in-0
          myPayloadConsumerFromTopic2-in-0: myPayloadConsumerFromTopic2-in-0

不过通常方案1更直接，能快速解决你的问题。

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