如何控制Spring Integration中首个Split的消息处理顺序？

要实现首个Split拆分出的第二条消息等待第一条处理完成（到达aggregateChannel）后再发送至inputChannel，可以通过暂存第二条消息+监听第一条处理完成信号触发发送的方式实现，具体步骤如下：

1. 核心思路 #

默认的Split组件会并行发送拆分后的所有消息到下游通道，我们需要：

• 利用Spring Integration自动生成的sequenceNumber消息头，区分拆分后的第一条和第二条消息；
• 将第一条消息直接发送到inputChannel处理，第二条消息暂存到队列通道；
• 监听aggregateChannel，当第一条消息的处理结果到达时，再将暂存的第二条消息发送到inputChannel。

2. 代码修改实现 #

2.1 定义暂存通道 #

创建队列通道用于暂存第二条消息：

@Bean
public MessageChannel secondMessageChannel() {
    return new QueueChannel();
}

2.2 修改首个Split的路由逻辑 #

在firstFlow中，拆分消息后根据sequenceNumber路由，第一条直接发inputChannel，第二条暂存：

@Bean
public IntegrationFlow firstFlow() {
    return IntegrationFlows.from("firstChannel")
            .split() // 自动添加sequenceNumber（从1开始）、sequenceSize等消息头
            .route(Message.class, msg -> {
                Integer seqNum = msg.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Integer.class);
                return seqNum == 1 ? "inputChannel" : "secondMessageChannel";
            })
            .get();
}

2.3 传递原消息标识到聚合结果 #

修改inputFlow，在处理mapping时将原消息的sequenceNumber保留到聚合后的消息头中，方便后续识别：

@Bean
public IntegrationFlow inputFlow() {
    return IntegrationFlows.from("inputChannel")
            .channel(MessageChannels.executor(taskExecutor()))
            .split()
            .handle((payload, headers) -> {
                // 执行mapping逻辑，并保留原消息的sequenceNumber
                Object mappedResult = mapping(payload);
                return MessageBuilder.withPayload(mappedResult)
                        .copyHeaders(headers)
                        .setHeader("originalSequenceNumber", headers.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
                        .build();
            })
            .aggregate()
            .channel("aggregateChannel")
            .get();
}

2.4 添加触发逻辑，发送第二条消息 #

创建Flow监听aggregateChannel，当第一条消息的聚合结果到达时，取出暂存的第二条消息发送到inputChannel：

@Bean
public IntegrationFlow triggerSecondMessageFlow() {
    return IntegrationFlows.from("aggregateChannel")
            // 过滤出第一条消息的聚合结果
            .filter(msg -> {
                Integer originalSeq = msg.getHeaders().get("originalSequenceNumber", Integer.class);
                return originalSeq != null && originalSeq == 1;
            })
            .handle((payload, headers) -> {
                // 从暂存通道取出第二条消息
                return secondMessageChannel().receive(1000);
            })
            // 将第二条消息发送到inputChannel处理
            .channel("inputChannel")
            .get();
}

3. 额外说明 #

• 该方案基于Spring Integration原生组件实现，无需额外依赖；
• 若拆分后的消息数量超过2条，可扩展路由和触发逻辑，实现按顺序依次触发；
• 暂存通道使用QueueChannel保证消息不丢失，若需持久化可替换为JmsChannel等持久化通道。

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