Spring Cloud Stream升级后@StreamListener与Consumer接口Kafka监听器行为差异问询

问题背景 #

我们正将Spring Boot从2.2.6.RELEASE升级至2.7.8，同时Spring Cloud Stream（SCS）从3.0.4.RELEASE升级至3.2.6.RELEASE，期间遇到Kafka监听器行为不一致的问题：

场景1：旧版本（@StreamListener）行为 #

旧版本使用@StreamListener，同一个TEST_CHANNEL绑定到TEST_TOPIC和TEST_TOPIC_GROUP消费组，两个@StreamListener会接收每条消息的副本并处理，但我们预期同一消费组下应轮询消费（同一时间仅一个监听器处理消息）。

代码示例：

@Component
public class TestListener1 {   

    @StreamListener("TEST_CHANNEL")
    public void handle(final Message<String> message) {
        log.info("execute Test Listener 1" );        
    }
}

@Component
public class TestListener2 {   

    @StreamListener("TEST_CHANNEL")
    public void handle(final Message<String> message) {
        log.info("execute Test Listener 2" );        
    }
}

配置示例：

spring:
  cloud:
    stream:
      bindings:
        TEST_CHANNEL:
          binder: kafka
          content-type: application/json
          destination: TEST_TOPIC
          group: TEST_TOPIC_GROUP

场景2：新版本（Consumer接口）行为 #

升级后@StreamListener已废弃，改用实现Consumer接口的监听器，创建两个不同通道TEST_CHANNEL_1、TEST_CHANNEL_2，均绑定到TEST_TOPIC和TEST_TOPIC_GROUP消费组，此时消息以轮询方式分发给两个监听器，每条消息仅被一个监听器处理，与场景1行为不同。

代码示例：

@Component("TEST_CHANNEL_1")
public class TestListener1 implements Consumer<Message<String>> {
    
    @Override
    public void accept(final Message<String> message) {
        log.info("execute Test Listener 1" );        
    }
}

@Component("TEST_CHANNEL_2")
public class TestListener2 implements Consumer<Message<String>> {

    @Override    
    public void accept(final Message<String> message) {
        log.info("execute Test Listener 2" );        
    }
}

配置示例：

spring:
  cloud:
    stream:
      bindings:
        TEST_CHANNEL_1:
          binder: kafka
          content-type: application/json
          destination: TEST_TOPIC
          group: TEST_TOPIC_GROUP
        TEST_CHANNEL_2:
          binder: kafka
          content-type: application/json
          destination: TEST_TOPIC
          group: TEST_TOPIC_GROUP

解决方案建议 #

1. 恢复旧版广播行为（需每条消息被所有监听器处理） #

如果需要回到旧版本中每个监听器都接收消息副本的广播效果，无需创建多个独立通道，利用Spring Cloud Stream的multiplex配置即可实现：

• 定义多个Consumer函数（替代@Component方式更适配函数式模型）：

@Component
public class TestListeners {
    @Bean
    public Consumer<Message<String>> testListener1() {
        return message -> log.info("execute Test Listener 1");
    }

    @Bean
    public Consumer<Message<String>> testListener2() {
        return message -> log.info("execute Test Listener 2");
    }
}

• 配置单个通道并绑定所有函数，开启multiplex=true：

spring:
  cloud:
    stream:
      function:
        bindings:
          testListener1-in-0: TEST_CHANNEL
          testListener2-in-0: TEST_CHANNEL
      bindings:
        TEST_CHANNEL:
          binder: kafka
          content-type: application/json
          destination: TEST_TOPIC
          group: TEST_TOPIC_GROUP
          consumer:
            multiplex: true

multiplex=true会让Spring Cloud Stream绕过Kafka消费组的分区分配逻辑，将消息广播给所有绑定的消费者，和旧版@StreamListener行为一致。

2. 保持轮询行为（符合Kafka原生设计） #

如果新版本的轮询行为符合预期（同一消费组下每条消息仅被一个监听器处理），当前实现是正确的。这是因为Kafka消费组的原生机制就是：同一消费组内的多个消费者会按分区轮询消费，每条消息只会被消费组内的一个实例处理。旧版本@StreamListener的广播行为其实是Spring Cloud Stream早期的特殊处理，并不符合Kafka的原生设计。

3. 差异原因说明 #

• 旧版本@StreamListener：同一个通道下的多个监听器默认以广播模式工作，所有监听器都会收到全量消息。
• 新版本函数式模型：每个Consumer对应独立的输入绑定，当多个绑定指向同一主题和消费组时，Kafka原生消费组机制生效，自动触发轮询消费。

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