Spring Integration聚合器负载类型错误：Entity而非List<Message<?>>

你编写的Spring Integration流程中，期望聚合器接收的负载类型为List<Message<?>>，但实际收到的是Entity，原因及解决方案如下：

核心原因 #

• 集合返回值的自动拆分机制：Spring Integration默认会对处理器返回的Collection（如List）进行自动拆分，将集合内的每个元素作为独立消息发送到下游通道。你的createCommonPackage方法返回List<Message<?>>后，框架会把列表中的每个Message单独拆分发送，而非将整个列表作为单条消息的负载传递，导致聚合器无法接收到完整的List<Message<?>>。
• 聚合逻辑配置缺失：当前流程仅通过split拆分了输入消息，但未针对publishSubscribeChannel的多分支结果配置对应聚合器来收集处理后的列表。若后续聚合器未关联分支处理结果，可能误收集到原拆分的Entity对象。

解决方案 #

1. 禁用集合自动拆分 #

通过注解或Flow配置，让框架保留返回的List<Message<?>>作为整体负载：

// 方式1：在处理器方法上添加注解
@Splitter(applySequence = false)
private List<Message<?>> createCsvPackage(Entity payload, MessageHeaders headers) {
    return createCommonPackage(payload, headers, flag);
}

// 方式2：在Flow中显式配置
.handle(this::createCsvPackage)
.split(Collections.singletonList(null), spec -> spec.applySequence(false)) // 禁用自动拆分，保留原列表
.channel(FTP_CHANNEL)

2. 添加显式聚合器 #

在FTP_CHANNEL下游配置聚合器，基于关联ID收集多分支结果：

.channel(FTP_CHANNEL)
.aggregate(spec -> spec
        .correlationStrategy(message -> message.getHeaders().get("correlationId"))
        .releaseStrategy(group -> group.getMessages().size() == 3) // 匹配订阅分支数量
        .outputProcessor(group -> group.getMessages().stream()
                .map(Message::getPayload)
                .flatMap(List::stream)
                .collect(Collectors.toList()))
)

3. 分支消息标识优化 #

为每个分支的消息添加专属头信息，帮助聚合器精准区分处理结果，避免原Entity流入聚合逻辑：

.subscribe(s -> s.filter(source -> filter1((Entity)source))
        .enrichHeaders(h -> h.header("packageType", "CSV"))
        .handle(this::createCsvPackage)
        .channel(FTP_CHANNEL)
)

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