如何用Spring Cloud Stream Kafka Binder实现批量消费后发布多条消息

你遇到的两个问题的核心原因分别是：

1. 返回List<Message<OutgoingEvent>>时输出单条数组消息：默认Spring Cloud Stream会把函数返回的集合当作单个消息的payload序列化，未开启输出拆分配置
2. 返回Flux时序列化异常：未引入响应式依赖或未正确配置响应式函数绑定，该场景无需使用响应式API即可实现

第一步：配置文件设置（application.yml） #

spring:
  cloud:
    stream:
      function:
        definition: aggregate # 绑定你定义的函数名
      kafka:
        binder:
          brokers: 你的kafka地址:9092
        bindings:
          aggregate-in-0: # 函数输入binding命名规则：<函数名>-in-<索引>
            consumer:
              batch-mode: true # 开启批量消费
              configuration:
                max.poll.records: 50 # 单次拉取最大消息数，可自定义
                fetch.min.bytes: 1024 # 批量拉取最小字节数，可自定义
      bindings:
        aggregate-in-0:
          destination: 输入topic名称
          group: 消费者组名称
          consumer:
            batch-mode: true
        aggregate-out-0: # 函数输出binding命名规则：<函数名>-out-<索引>
          destination: 输出topic名称
          producer:
            split-output: true # 核心配置：将返回的List拆分为单条消息分别发送

第二步：正确代码实现 #

import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.kafka.support.KafkaHeaders;
import java.util.List;
import java.util.Map;
import java.util.ArrayList;
import java.util.stream.Collectors;
import java.util.function.Function;

public class AggregateProcessor {

    public Function<List<IncomingEvent>, List<Message<OutgoingEvent>>> aggregate() {
        return batch -> {
            // 修正原代码中分组写法错误
            Map<Key, List<IncomingEvent>> resultsMap = batch.stream()
                    .collect(Collectors.groupingBy(IncomingEvent::key));
            List<Message<OutgoingEvent>> aggregationResults = new ArrayList<>();
            
            for (Map.Entry<Key, List<IncomingEvent>> resultGroup : resultsMap.entrySet()) {
                OutgoingEvent outgoingEvent = getOutgoingEvent(resultGroup.getKey(), resultGroup.getValue());
                aggregationResults.add(
                        MessageBuilder.withPayload(outgoingEvent)
                                .setHeader(KafkaHeaders.MESSAGE_KEY, outgoingEvent.getKey())
                                .build()
                );
            }
            return aggregationResults;
        };
    }

    // 你的聚合逻辑实现
    private OutgoingEvent getOutgoingEvent(Key key, List<IncomingEvent> events) {
        // 自定义聚合逻辑
        return new OutgoingEvent(key, events);
    }

    // 示例Key、IncomingEvent、OutgoingEvent定义可根据业务自行修改
    public record Key(String id) {}
    public record IncomingEvent(Key key, String content) {}
    public record OutgoingEvent(Key key, List<IncomingEvent> events) {
        public Key getKey() {
            return key;
        }
    }
}

注意事项 #

• 无需修改返回类型为Flux，配合split-output: true配置，返回的List会自动拆分为独立消息发送，避免序列化问题
• 确保输入binding的批量配置两处都开启（kafka.binder下和bindings下的consumer.batch-mode都设为true）
• 原代码中分组逻辑的lambda写法错误已修正，避免运行时类型转换异常

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