Spring Cloud Stream Kafka消费者批量处理不生效，如何配置批量接收？

问题背景 #

使用Spring Boot 3.2.9 + Spring Cloud 2023.0.3开发，期望Kafka消费者一次性接收多条消息组成的列表，但实际每次仅收到单条消息，需要调整配置和代码实现批量处理。

当前代码与配置 #

主应用代码 #

@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

    @Bean
    public Function<Message<List<String>>, String> route() {
        return input -> {
            System.out.println("processing: " + input);
            return input.getPayload() + " response";
        };
    }
}

核心配置（application.yaml） #

spring:
  application:
    name: demo
  cloud:
    function:
      definition: route
    stream:
      bindings:
        route-out-0:
          destination: out-topic
        route-in-0:
          destination: in-topic
          group: group1
          consumer:
            batch-mode: true

测试环境配置（application-local.yaml） #

spring:
  cloud:
    stream:
      bindings:
        test-out-0:
          destination: in-topic

测试代码 #

@SpringBootTest
@EmbeddedKafka
@ActiveProfiles("local")
@Import(TestChannelBinderConfiguration.class)
class DemoApplicationTests {

    @Autowired
    StreamBridge streamBridge;

    @Autowired
    OutputDestination outputDestination;


    @Test
    void test() throws InterruptedException {
        System.out.println("sending");
        streamBridge.send("test-out-0", "some message1".getBytes());
        streamBridge.send("test-out-0", "some message2".getBytes());
        streamBridge.send("test-out-0", "some message3".getBytes());
        System.out.println("received: " + new String(outputDestination.receive(1000, "out-topic").getPayload()));
        System.out.println("received: " + new String(outputDestination.receive(1000, "out-topic").getPayload()));
        System.out.println("received: " + new String(outputDestination.receive(1000, "out-topic").getPayload()));
    }

}

问题现象 #

实际运行日志显示消费者被调用3次，且消息未反序列化为String：

processing: GenericMessage [payload=byte[13], headers={source-type=kafka, id=8d2ff990-91d1-dde3-1874-525036375f80, contentType=application/json, timestamp=1725719498271, target-protocol=kafka}]
processing: GenericMessage [payload=byte[13], headers={source-type=kafka, id=305a409e-5e43-7c58-d1dd-215d3595f56d, contentType=application/json, timestamp=1725719498271, target-protocol=kafka}]
processing: GenericMessage [payload=byte[13], headers={source-type=kafka, id=cc6a41a3-9ffe-19cc-71ef-a211314f2e5f, contentType=application/json, timestamp=1725719498287, target-protocol=kafka}]

若将Function<Message<List<String>>, String>改为Function<Message<String>, String>，消费者仍被调用3次，但能正常接收String类型消息。

解决方案 #

1. 修正函数泛型定义 #

Spring Cloud Stream的批量模式下，框架会将多条消息包装为**List**传递给函数，而非将多条消息的payload合并到单个Message中。因此需要调整Function的泛型：

仅处理消息payload的写法

@Bean
public Function<List<String>, String> route() {
    return input -> {
        System.out.println("processing batch: " + input);
        return input.toString() + " response";
    };
}

需要处理消息头的写法

@Bean
public Function<List<Message<String>>, String> route() {
    return input -> {
        List<String> payloads = input.stream()
                                     .map(Message::getPayload)
                                     .toList();
        System.out.println("processing batch with headers: " + payloads);
        return payloads.toString() + " response";
    };
}

2. 补充Kafka批量消费配置 #

仅开启batch-mode: true不够，还需配置Kafka消费者的批量拉取参数，确保能积累足够消息再返回：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          route-in-0:
            consumer:
              max-poll-records: 100 # 每次拉取的最大消息数
              fetch-min-size: 3      # 至少积累3条消息才返回（匹配测试发送的条数）
              fetch-max-wait: 5000   # 最多等待5秒，超时后即使条数不足也返回
              auto-commit-interval: 1000 # 自动提交offset的间隔
      bindings:
        route-in-0:
          destination: in-topic
          group: group1
          consumer:
            batch-mode: true
            content-type: text/plain # 指定消息类型，确保String反序列化生效

3. 调整测试代码（可选） #

测试时可增加短暂延迟，让Kafka有时间积累消息：

@Test
void test() throws InterruptedException {
    System.out.println("sending");
    streamBridge.send("test-out-0", "some message1".getBytes());
    streamBridge.send("test-out-0", "some message2".getBytes());
    streamBridge.send("test-out-0", "some message3".getBytes());
    
    // 等待Kafka积累消息
    Thread.sleep(1000);
    
    // 此时只需要接收一次，因为批量处理后只会返回一条响应
    System.out.println("received batch response: " + new String(outputDestination.receive(2000, "out-topic").getPayload()));
}

原理说明 #

• 原代码泛型错误：Message<List<String>>不符合Spring Cloud Stream批量模式的参数约定，框架无法正确解析批量消息，因此降级为单条处理，且因泛型不匹配导致反序列化失败。
• Kafka批量拉取依赖fetch-min-size和fetch-max-wait参数，控制消费者何时返回拉取到的消息，确保能攒够指定条数或超时后批量返回。

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