Spring Integration多线程HTTP请求聚合：仅最后请求获回复其余超时问题

问题背景 #

搭建了Spring Boot + Spring Integration应用实现HTTP请求负载聚合批量处理，核心组件包括：

• RestController：接收HTTP请求负载
• MessagingGateway：将请求发送到通道，等待聚合处理后的回复
• IntegrationFlow：基于correlationId和SequenceSizeReleaseStrategy聚合消息
• 自定义消息组处理器：批量处理聚合后的消息并返回结果

问题现象 #

设置序列大小为3，连续发送3个请求后：

• 请求1、2：网关超时，返回null
• 请求3：成功收到聚合处理后的结果

核心原因 #

Spring Integration聚合器默认仅将聚合结果发送给触发组释放的那个请求的回复通道，前两个请求的replyChannel头信息未被聚合器保留，导致它们的网关一直等待直到超时。即使将responseQueue改为发布订阅通道也无效，因为每个网关请求的回复通道是临时生成的，并非共用的固定通道。

解决方案 #

1. 开启聚合器的sendRepliesToAll属性 #

修改聚合器配置，添加.sendRepliesToAll(true)，让聚合器将结果发送给组内所有消息的回复通道：

@Bean
public IntegrationFlow myAggregatorFlow(MyGroupHandler myGroupHandler) {
    return IntegrationFlow
            .from(requestQueue())
            .aggregate(a -> a
                    .releaseStrategy(new SequenceSizeReleaseStrategy())
                    .expireGroupsUponCompletion(true)
                    .sendPartialResultOnExpiry(true)
                    .sendRepliesToAll(true)) // 开启给所有请求发送回复
            .handle(myGroupHandler)
            .channel("responseQueue")
            .get();
}

2. 移除网关的固定replyChannel配置（关键） #

原网关中指定的固定replyChannel = "responseQueue"会导致所有请求共享同一个回复通道，但每个同步网关调用都会生成自己的临时回复通道来匹配请求。移除该配置，让Spring自动管理临时回复通道：

@MessagingGateway
public interface MyIntegrationGateway {

    @Gateway(requestChannel = "requestQueue") // 移除固定replyChannel
    String sendToFlow(MyDataObject payload,
                      @Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String correlationId,
                      @Header(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE) int groupSize) throws MessagingException;

}

3. 验证依赖版本 #

sendRepliesToAll属性从Spring Integration 5.0版本开始支持，请确保你的spring-integration-core依赖版本≥5.0。

效果验证 #

修改完成后，连续发送3个请求，所有请求都会在聚合组处理完成后收到相同的结果，不会出现超时情况。

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