Spring Integration Scatter-Gather聚合响应报错及实现方法问询

一、报错排查与修复 #

1. 「无法访问payload属性」报错修复 #

• 核心原因：HTTP出站网关默认返回ResponseEntity<Dummy>而非直接的Dummy对象，后续组件尝试访问payload属性时找不到对应字段；或消息传递过程中payload被意外包装。
• 修复方案：
  • 配置HTTP出站网关直接提取响应体作为消息payload：

    @Bean
    public MessageHandler dummyHttpGateway1() {
        return Http.outboundGateway("http://localhost:8080/api/dummy1")
                .httpMethod(HttpMethod.GET)
                .expectedResponseType(Dummy.class)
                .extractPayload(true) // 关键配置：跳过ResponseEntity，直接返回Dummy对象
                .get();
    }
    

  • 若已返回ResponseEntity，添加转换器提取body：

    @Bean
    @Transformer(inputChannel="postHttpChannel", outputChannel="scatterInputChannel")
    public Function<Message<ResponseEntity<Dummy>>, Dummy> extractResponseBody() {
        return msg -> msg.getPayload().getBody();
    }
    

2. 「无返回结果」报错修复 #

• 核心原因：自定义释放策略逻辑错误、消息分组关联失效、聚合超时时间过短。
• 修复方案：
  • 确保释放策略匹配接口数量（3个）：

    @Bean
    public ReleaseStrategy dummyReleaseStrategy() {
        return messageGroup -> messageGroup.size() == 3; // 当3个响应全部到达时触发释放
    }
    

  • 调整聚合超时时间，避免提前终止：

    @Bean
    public ScatterGatherHandler scatterGatherHandler() {
        ScatterGatherHandler handler = new ScatterGatherHandler(scatterChannel(), gatherChannel());
        handler.setGatherTimeout(15000); // 设为15秒，根据实际接口响应时间调整
        return handler;
    }
    

  • 禁止手动修改correlationId消息头，Scatter-Gather会自动维护分组关联，手动修改会导致分组错误。

二、多Dummy对象聚合逻辑实现 #

1. 自定义聚合处理器 #

继承AbstractAggregatingMessageGroupProcessor，根据业务需求合并属性（示例：合并多个Dummy的dataList字段）：

public class DummyAggregator extends AbstractAggregatingMessageGroupProcessor {
    @Override
    protected Object aggregatePayloads(MessageGroup group, Map<String, Object> headers) {
        List<Dummy> dummyList = group.getMessages().stream()
                .map(msg -> (Dummy) msg.getPayload())
                .collect(Collectors.toList());

        Dummy aggregatedDummy = new Dummy();
        aggregatedDummy.setId("merged-" + UUID.randomUUID());
        List<String> mergedData = new ArrayList<>();
        dummyList.forEach(d -> mergedData.addAll(d.getDataList()));
        aggregatedDummy.setDataList(mergedData);

        return aggregatedDummy;
    }
}

2. 配置聚合器 #

将自定义处理器绑定到聚合通道：

@Bean
@ServiceActivator(inputChannel="gatherChannel")
public AggregatingMessageHandler aggregatorHandler() {
    AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new DummyAggregator());
    aggregator.setReleaseStrategy(dummyReleaseStrategy());
    aggregator.setCorrelationStrategy(new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID));
    aggregator.setOutputChannel("aggregatedResultChannel");
    return aggregator;
}

3. 验证聚合结果 #

在测试中监听结果通道，确认聚合效果：

@Autowired
private MessageChannel aggregatedResultChannel;

@Test
void testAggregation() {
    Message<String> triggerMsg = MessageBuilder.withPayload("start").build();
    aggregatedResultChannel.subscribe(msg -> {
        Dummy merged = (Dummy) msg.getPayload();
        Assertions.assertEquals(3, merged.getDataList().size()); // 假设每个Dummy返回1条数据，合并后共3条
    });
    scatterGatherHandler().handleMessage(triggerMsg);
}

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