Spring Integration Aggregator 结合 Publish Subscriber答案

【问题标题】：Spring Integration Aggregator combined with Publish SubscriberSpring Integration Aggregator 结合 Publish Subscriber
【发布时间】：2016-06-16 19:04:51
【问题描述】：

我有关于 Spring 集成的基本问题。我尝试在其他线程中搜索，但对答案不满意。所以这是我的问题。任何提示将不胜感激。

我正在尝试从 4 个不同来源并行获取数据并聚合数据。

我的理解（根据我之前阅读的线程之一）是，如果拆分器的任何输出通道是直接通道，那么请求将按顺序路由。

但是 Splitter 的输出通道是一个接收列表路由器。当我调试请求时，请求是按顺序路由的。所以经过一些分析，我发现我们可以使用有效负载类型的路由器来并行处理。

所以我将路由器更改为有效负载类型的路由器，但请求仍然是按顺序路由的。

所以我改变了我的策略，使用了一个发布订阅频道，并将所有订阅频道的输出连接到一个聚合器，以便我可以聚合结果。我能够并行获取数据，但我没有看到响应被路由到聚合。我做错了吗？请提出建议！

    <int:publish-subscribe-channel id="PublishSubscriberChannel"
    task-executor="taskExecutor" />

<task:executor id="taskExecutor" pool-size="5" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="getVehicleInfoWithAdditionalAttributes"
    output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" />

【问题讨论】：

标签： spring-integration splitter

【解决方案1】：

所以最后，我发现我做错了什么。最初，我正在研究 Spring Integration 2.0 版。这是一个旧应用程序。因此，当我的代码正是问题中给出的内容时，我看不到请求被路由到聚合器。然后，当我将 Spring 版本升级到 4.3.2 RELEASE 时，我开始收到 Correlation Strategy 可能为空的错误。那是当我发现聚合器需要关联 ID 并且如果我自动添加了拆分器关联 ID 并且对于发布订阅通道，我们需要添加 apply-sequence="true" 以添加默认关联 ID，它可以由聚合器使用。所以从本质上讲，这就是我的代码现在的样子，并且运行良好。

<int:publish-subscribe-channel id="PublishSubscriberChannel" apply-sequence="true"
    task-executor="taskExecutor" />

<task:executor id="taskExecutor" pool-size="5" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource3"
    output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" />

【讨论】：