Spring Cloud Stream单函数绑定多Kafka主题与消费组配置问询

该需求可以实现，但你当前的配置写法存在语法错误，无法生效。
Spring Cloud Stream 中单个 input 绑定仅支持配置1个消费组，不支持通过逗号分隔给同一个绑定配置多个消费组。要实现单函数对接多主题+多独立消费组的需求，需要通过「同函数多input绑定」的方式实现，框架会自动将所有绑定收到的消息路由到同一个消费函数处理。

你原来的配置存在两个核心错误：

• group 配置项不支持逗号分隔传多个值，传入多个值时会被整体识别为一个非法的消费组名称字符串，不会解析为三个独立消费组
• 逗号分隔配置destination的写法仅适用于同一个消费组消费多个主题的场景，无法满足三个主题对应三个独立消费组的要求

application.yml 配置写法 #

spring:
  cloud:
    function:
      # 仅需声明一次目标函数，末尾无需多余分号
      definition: handleEvent
    stream:
      bindings:
        # 按 函数名-in-序号 规则定义多个独立绑定，序号从0开始递增
        handleEvent-in-0:
          destination: topicAEvent
          group: AEventConsumergroup
        handleEvent-in-1:
          destination: topicBEvent
          group: BEventConsumergroup
        handleEvent-in-2:
          destination: topicCEvent
          group: CEventConsumerGroup
      kafka:
        bindings:
          # 如需为每个消费组单独配置Kafka专属参数，在此处对应配置即可
          handleEvent-in-0:
            consumer:
              auto-commit-offset: true
          handleEvent-in-1:
            consumer:
              auto-commit-offset: true
          handleEvent-in-2:
            consumer:
              auto-commit-offset: true

函数编写注意事项 #

• 消费函数无需做特殊改造，正常编写业务逻辑即可，三个绑定的消息都会自动投递到该函数
• 如果需要区分消息来源主题，可以在函数参数中传入Kafka原生ConsumerRecord对象，调用其topic()方法即可获取当前消息所属的主题名，示例代码：

@Bean
public Consumer<ConsumerRecord<String, String>> handleEvent() {
    return record -> {
        String sourceTopic = record.topic();
        String messageValue = record.value();
        // 编写统一业务处理逻辑
    };
}

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