如何聚合来自多个渠道的响应

Question

我有一个带有以下功能的弹簧集成实现：

1. 多个发布渠道在一个公共渠道上发布。
2. 所有通道都返回相同的响应对象。
3. 聚合器试图聚合来自上述所有渠道的响应

问题：聚合器无法将所有响应组合在一起，并且在通道的第一个响应时调用提供的方法

这里是详细信息。我需要做什么来汇总回复？

<int:publish-subscribe-channel id="aggregate-channel" apply-sequence="true"/>
<int:publish-subscribe-channel id="input-channel" apply-sequence="true"/>

<int:service-activator input-channel="input-channel" output-channel="aggregate-channel" ref="...A" method="...A">
<int:service-activator input-channel="input-channel" output-channel="aggregate-channel" ref="...B" method="...B">
<int:service-activator input-channel="input-channel" output-channel="aggregate-channel" ref="...C" method="...C">
<int:service-activator input-channel="input-channel" output-channel="aggregate-channel" ref="...D" method="...D">

<!--This is the aggregator. 
**Expecting a list of size 4 but then it gets list of size 1 for each response channel
-->
<int:aggregator input-channel="aggregate-channel" output-channel="gateway-response-channel" ref="Service" method="responseListProcessor"/>

Answer 1

解决方案

• 替换

<int:publish-subscribe-channel id="aggregate-channel" apply-sequence="true"/>

    with

<int:publish-subscribe-channel id="aggregate-channel"/>

问题

• 最初它以序列大小4 开始，因为input-channel 有4 订阅者
• 但是当您将属性apply-sequence="true" 添加到aggregate-channel 时，它会将序列大小重置为1，因为aggregate-channel 只有一个订阅者，即aggregator。

参考

https://docs.spring.io/spring-integration/docs/5.1.7.RELEASE/reference/html/#channel-configuration-pubsubchannel

如果您在 PublishSubscribeChannel 下游提供聚合器，则可以将通道上的“apply-sequence”属性设置为 true。

这样做表示通道应该在传递消息之前设置序列大小和序列号消息头以及相关 ID。

例如，如果有五个订阅者，则序列大小将设置为 5，并且消息的序列号标头值范围为 1 到 5。

Answer 2

我会说您的要求已完全满足此类任务的特定 EIP - Scatter-Gather：https://docs.spring.io/spring-integration/docs/current/reference/html/message-routing.html#scatter-gather