Spring Integration拆分、路由、聚合逻辑失效，子流处理后流程中断如何解决？

核心原因分析 #

你的流程在子流处理后停止，主要有两个关键问题：

1. 关联ID生成错误：当前在split后为每个消息生成独立的CORRELATION_ID，导致聚合器无法将同批次的子消息归为一组，会一直等待所有分组消息（永远无法满足），流程卡在聚合环节。
2. 子流未传递消息：如果firebaseSender.sendPush等方法返回void，子流的handle会成为终端操作，消息不会传递回主流程的聚合器节点。

解决方案 #

1. 提前生成关联ID #

在split之前生成全局关联ID，确保拆分后的所有子消息继承同一个CORRELATION_ID，让聚合器能正确识别同批次消息。

2. 确保子流传递消息 #

• 修改推送处理器（FirebaseSender等）的sendPush方法，返回处理后的PushNotification或结果对象；
• 若无法修改处理器，可在子流的handle后添加.bridge()，强制将消息传递到下游。

3. 配置聚合器释放策略 #

添加明确的释放策略，避免聚合器无限等待（比如基于消息数量、超时时间）。

修正后的代码 #

public class PushSendFlowAdapter extends IntegrationFlowAdapter {

    private final String inputChannel;
    private final DeviceTokenManagerService<?> deviceTokenManagerService;
    private final FirebaseSender firebaseSender;
    private final ApnsSender apnsSender;
    private final HuaweiSender huaweiSender;
    private static final Logger log = LoggerFactory.getLogger(PushSendFlowAdapter.class);

    @Override
    protected IntegrationFlowDefinition<?> buildFlow() {
        return IntegrationFlows.from(this.inputChannel)
                // 1. 先为原始消息生成全局关联ID，split后的子消息会继承该ID
                .enrichHeaders(headerEnricherSpec -> 
                        headerEnricherSpec.header(TracingConstants.CORRELATION_ID, UUID.randomUUID()))
                // 复制通知到每个设备令牌
                .handle((pushNotification, headers) -> 
                        deviceTokenManagerService.clonePushNotificationForEachDeviceToken((PushNotification) pushNotification))
                // 拆分消息列表
                .split()
                //  enrich平台信息到header
                .enrichHeaders(headerEnricherSpec -> 
                        headerEnricherSpec.headerExpression(TracingConstants.PLATFORM, "payload.getDeviceToken().getPlatform()"))
                // 路由到对应平台子流
                .route(Message.class, h -> h.getHeaders().get(TracingConstants.PLATFORM, Platform.class),
                        routerSpec -> routerSpec
                                .subFlowMapping(Platform.ANDROID, androidFlow -> androidFlow
                                        .handle(PushNotification.class, firebaseSender::sendPush)
                                        // 若sendPush返回void，添加bridge确保消息传递
                                        .bridge())
                                .subFlowMapping(Platform.IOS, iosFlow -> iosFlow
                                        .handle(PushNotification.class, apnsSender::sendPush)
                                        .bridge())
                                .subFlowMapping(Platform.HUAWEI, huaweiFlow -> huaweiFlow
                                        .handle(PushNotification.class, huaweiSender::sendPush)
                                        .bridge())
                )
                // 聚合同关联ID的所有子消息
                .aggregate(aggregatorSpec -> aggregatorSpec
                        .correlationStrategy(message -> message.getHeaders().get(TracingConstants.CORRELATION_ID))
                        // 释放策略：当所有子消息都到达时释放（基于split的原始消息数量）
                        .releaseStrategy(group -> group.getMessages().size() == group.getSequenceSize())
                        // 超时兜底：避免因消息丢失导致无限等待
                        .groupTimeout(5000))
                // 处理聚合结果
                .handle((p, h) -> {
                    log.info("聚合结果日志: {}", p);
                    // 业务逻辑处理
                    return null;
                });
    }
}

额外说明 #

• 如果你的sendPush方法本身会返回处理后的消息（比如带推送状态的PushNotification），可以去掉子流中的.bridge()；
• groupTimeout可根据业务场景调整，确保在异常情况下流程不会永久阻塞；
• split组件会自动为子消息添加sequenceNumber和sequenceSize头，因此releaseStrategy可以直接使用group.getMessages().size() == group.getSequenceSize()来判断是否所有子消息都已处理完成。

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