Spring Integration拆分后聚合异常：无Address的AppDetails致流程阻塞

问题根源 #

当AppDetails的Address列表为空时，并行调用环节没有消息产生，但聚合器仍在等待预设数量的消息完成聚合，导致整个流程阻塞。聚合器默认不会主动处理空消息组，必须收到足够数量的消息才会释放结果。

解决方案 #

1. 前置空值判断，短路无效流程

在进入并行调用前，直接判断Address是否为空，若为空则直接返回空数组，跳过后续的拆分、并行调用和聚合逻辑，从根源避免聚合器等待。

2. 给聚合器配置超时兜底

即使有前置判断，仍建议给聚合器添加超时配置，防止其他异常场景导致的阻塞。通过groupTimeout设置超时时间，expireGroupsUponCompletion确保组完成后自动清理。

3. 调整聚合器的释放策略

确保释放策略能兼容空消息组的情况，或者在Address为空时主动发送一个空标记消息，让聚合器能正常完成。

修改后的核心代码示例 #

@Bean
public IntegrationFlow flow3() {
    return f -> f
            .<AppDetails, Object>transform(payload -> {
                // 前置判断Address是否为空，直接返回空数组
                if (payload.getAddresses() == null || payload.getAddresses().isEmpty()) {
                    return Collections.emptyList();
                }
                // 给原始 payload 打标，方便聚合时关联
                return MessageBuilder.withPayload(payload.getAddresses())
                        .setHeader("originalAppDetails", payload)
                        .build();
            })
            // 仅当 payload 不是空数组时才进入拆分并行流程
            .filter(payload -> !(payload instanceof List<?> && ((List<?>) payload).isEmpty()))
            .split()
            .channel(c -> c.executor(Executors.newFixedThreadPool(4))) // 并行线程池
            .handle((address, headers) -> {
                // 调用外部API，返回对应响应
                return apiGateway.fetchAddressDetails((Address) address);
            })
            .aggregate(a -> a
                    .correlationStrategy(m -> m.getHeaders().get("correlationId"))
                    .releaseStrategy(g -> {
                        AppDetails original = (AppDetails) g.getGroupMetadata().get("originalAppDetails");
                        return g.size() == original.getAddresses().size();
                    })
                    .groupTimeout(Duration.ofSeconds(2)) // 超时2秒自动释放
                    .expireGroupsUponCompletion(true)
                    .outputProcessor(g -> {
                        AppDetails original = (AppDetails) g.getGroupMetadata().get("originalAppDetails");
                        // 按原始Address顺序关联API响应
                        return original.getAddresses().stream()
                                .map(address -> {
                                    Optional<Object> response = g.getMessages().stream()
                                            .filter(msg -> address.equals(msg.getPayload()))
                                            .findFirst()
                                            .map(Message::getPayload);
                                    // 构造关联后的结果对象
                                    return response.map(res -> new AppAddressDetail(address, res))
                                            .orElse(null);
                                })
                                .collect(Collectors.toList());
                    })
            )
            // 处理空数组的情况，确保输出统一
            .transform(payload -> payload instanceof List<?> ? payload : Collections.emptyList());
}

测试验证 #

针对包含无Address的AppDetails的测试输入，比如：

[
  {
    "appId": "1",
    "addresses": [{"id": "a1"}, {"id": "a2"}]
  },
  {
    "appId": "2",
    "addresses": []
  }
]

修改后的流程会直接给appId=2的AppDetails返回空数组，不会进入并行和聚合环节，彻底解决阻塞问题。

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