Spring Integration流中多个ScatterGather如何实现顺序执行？

业务需求 #

需要调用A、B、C共3个API，其中A和B可并行执行，但C依赖B的输出结果，最终需要聚合全部3个服务的响应。

原有实现代码 #

public IntegrationFlow mainFlow()
{
    return flow -> flow
            .scatterGather(scatterRequests(), aggregateResponse())
            .enrichHeaders(s -> s.headerExpressions(h -> h.put(AGGREGATED_RESPONSE, "payload")))
            .scatterGather(scatterOtherRequests(), aggregateResponse());
}

private Consumer<RecipientListRouterSpec> scatterRequests()
{
    return scatterer -> scatterer
        .applySequence(true)
        .recipientFlow(flowA())
        .recipientFlow(flowB());
}

private Consumer<RecipientListRouterSpec> scatterOtherRequests()
{
    return scatterer -> scatterer
        .applySequence(true)
        .recipientFlow(flowC())
        .recipientFlow(aggregatedResponseSubFlow());
}

private IntegrationFlow raggregatedResponseSubFlow() 
{
     return subflow -> subflow
             .handle((payload, headers) -> {
                 return headers.get(AGGREGATED_RESPONSE);
             });
}

存在的问题 #

流中先后定义的两个ScatterGather出现了并行执行的情况，导致返回结果不一致。 #

Spring Integration的scatterGather组件默认如果没有显式配置同步执行，在子流配置了异步执行器的场景下，会出现第一个scatterGather还未完成聚合就向下游发送消息的情况，导致第二个scatterGather提前触发，读取到未赋值的AGGREGATED_RESPONSE头，最终结果异常。 #

要让第一个scatterGather完全执行结束后再启动第二个，只需要调整第一个scatterGather的配置，禁用异步转发，强制gather阶段同步阻塞到所有子流返回、聚合完成再向下游传递消息即可。

修改后的主流程代码： #

public IntegrationFlow mainFlow()
{
    return flow -> flow
            // 第一个scatterGather配置显式同步规则
            .scatterGather(scatterRequests(), 
                aggregator -> aggregator
                    .requireSequence(true)
                    .correlationStrategy(message -> message.getHeaders().getId())
                    .releaseStrategy(group -> group.size() == 2), // 明确等待A、B两个结果返回后再聚合
                // 核心配置：禁用scatterGather的异步转发，强制聚合完成后再向下游传递消息
                scatterGatherSpec -> scatterGatherSpec.async(false)
            )
            .enrichHeaders(s -> s.headerExpressions(h -> h.put(AGGREGATED_RESPONSE, "payload")))
            .scatterGather(scatterOtherRequests(), 
                aggregator -> aggregator
                    .requireSequence(true)
                    .releaseStrategy(group -> group.size() == 2)
            );
}

原有实现多了一次不必要的scatterGather操作，可简化流程，减少冗余的header传递和聚合逻辑，从根源上避免多scatterGather的执行顺序问题：

1. 第一步并行调用A和B，拿到A、B的聚合响应
2. 第二步用B的响应作为入参调用C，同时保留第一步的A、B结果
3. 第三步直接聚合A、B、C三个响应即可

简化后的代码示例： #

public IntegrationFlow mainFlow() {
    return flow -> flow
            // 第一步并行执行A、B
            .scatterGather(scatterRequests(), 
                aggregator -> aggregator.releaseStrategy(group -> group.size() == 2),
                spec -> spec.async(false)
            )
            // 存储A、B的聚合结果到header，同时提取B的结果作为调用C的入参
            .enrichHeaders(s -> s.headerExpression("AB_RESULT", "payload"))
            // 提取B的结果（索引可根据实际聚合的顺序调整）
            .<List<Object>, Object>transform(payload -> payload.get(1))
            // 调用C服务
            .gateway(flowC())
            // 聚合A、B、C三个服务的结果
            .handle((payload, headers) -> {
                List<Object> abResult = headers.get("AB_RESULT", List.class);
                Object aResponse = abResult.get(0);
                Object bResponse = abResult.get(1);
                Object cResponse = payload;
                // 自定义聚合逻辑
                return customAggregate(aResponse, bResponse, cResponse);
            });
}

private Consumer<RecipientListRouterSpec> scatterRequests() {
    return scatterer -> scatterer
            .applySequence(true)
            .recipientFlow(flowA())
            .recipientFlow(flowB());
}

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