有没有办法在 Splitter Aggregator 的服务激活器中设置超时？

Question

我创建了一个拆分器聚合器配置，它将拆分消息，每条消息将路由到特定的服务激活器。在每个服务激活器内部，它将调用一个网关，该网关使用 CXF 调用一个 SOAP 服务。我尝试将回复超时放在 SOAP 网关中，但它不起作用，仍然继续等待服务并聚合消息。我现在在我的代码中所做的是将回复超时放在消息网关中。

但问题是，如果一项服务因超时而失败，则将发送到聚合器的其余消息也会失败，因为它们属于同一个消息组。

我也尝试在我的聚合器中添加 send-partial-result-on-expiry="true" 但它仍然返回超时错误。

有没有办法在通道或服务激活器中设置超时？还是在 SOAP 网关中？这样如果一条消息超时失败，它不会影响成功？

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns=".....">

<context:component-scan base-package="com.t460.wilkins.services.impl"/>


<!--The gateways and service activators that will call a designated SOAP service using CXF-->
<int:gateway id="FirstGateway" default-request-channel="firstChannel" service-interface="com.t460.wilkins.gateways.FirstGateway"/>    
<int:gateway id="SecondGateway" default-request-channel="secondChannel" service-interface="com.t460.wilkins.gateways.SecondGateway"/>
<int:gateway id="ThirdGateway" default-request-channel="thirdChannel" service-interface="com.t460.wilkins.gateways.ThirdGateway"/>

<int:channel id="firstChannel"/>
<int:service-activator input-channel="firstChannel" ref="firstService" />
<int:channel id="secondChannel"/>
<int:service-activator input-channel="secondChannel" ref="secondService" />
<int:channel id="thirdChannel"/>
<int:service-activator input-channel="thirdChannel" ref="thirdService" />



<!--Configuration for the Splitter Aggregator-->
<!-- The gateway that will invoke the Splitter Aggregator. The gateway that will pass the initial message
and gather the aggregated message-->
<int:gateway id="MessageGateway" service-interface="com.t460.wilkins.gateways.MessageGateway" default-request-channel="asyncSenderChannel" default-reply-channel="asyncSenderChannel" default-reply-timeout="5000"/>

<int:channel id="asyncSenderChannel"/>
<int:channel id="asyncReceiverChannel"/>

<!-- Splitter-->
<int:splitter input-channel="asyncSenderChannel" output-channel="routingChannel" id="messageSplitter" ref="messageSplitter" />
<int:channel id="routingChannel"/>

<!-- Router -->
<int:recipient-list-router id="recipientWithSelector" input-channel="routingChannel">
    <int:recipient channel="firstSplitChannel" selector-expression="headers.msgType eq 'First'"/>
    <int:recipient channel="secondSplitChannel" selector-expression="headers.msgType eq 'Second'"/>
    <int:recipient channel="thirdSplitChannel" selector-expression="headers.msgType eq 'Third'"/>
</int:recipient-list-router>

<int:channel id="firstSplitChannel">
    <int:queue/>
</int:channel>
<int:channel id="secondSplitChannel">
    <int:queue/>
</int:channel>
<int:channel id="thirdSplitChannel">
    <int:queue/>
</int:channel>

<!-- These are the service activators where the splitted messages will be routed. Inside their classes, they each invoke the
an appropriate gateway listed above to get a data from a SOAP service using cxf-->
<int:service-activator input-channel="firstSplitChannel" output-channel="aggregateChannel" 
ref="firstSoapActivator">
    <int:poller receive-timeout="1000" task-executor="taskExecutor" fixed-rate="5"/>
</int:service-activator>

<int:service-activator input-channel="secondSplitChannel" output-channel="aggregateChannel" 
ref="secondSoapActivator">
    <int:poller receive-timeout="1000" task-executor="taskExecutor" fixed-rate="5"/>
</int:service-activator>

<int:service-activator input-channel="thirdSplitChannel" output-channel="aggregateChannel" 
ref="thirdSoapActivator">
    <int:poller receive-timeout="1000" task-executor="taskExecutor" fixed-rate="5"/>
</int:service-activator>

<!--Aggregator-->
<int:channel id="aggregateChannel"/>
<int:aggregator input-channel="aggregateChannel" output-channel="asyncReceiverChannel" id="aggregator"
ref="componentsAggregator" correlation-strategy="componentsCorrelationStrategy"
release-strategy="componentsReleaseStrategy" expire-groups-upon-completion="true" send-partial-result-on-expiry="true"/>

<!--Task Executor-->
<task:executor id="taskExecutor" pool-size="10-1000"
               queue-capacity="5000"/>

---更新----

我尝试删除消息网关的回复超时，并将 send-timeout="5000" 放在服务激活器中，但聚合器仍在等待所有消息到达。

我还尝试在 SOAP 网关“FirstGateway”、“SecondGateway”、“ThirdGateway”上设置回复超时，但它仍在推送并等待所有消息。

Answer 1

首先您需要考虑使用 Scatter-Gather：https://docs.spring.io/spring-integration/docs/5.0.8.RELEASE/reference/html/messaging-routing-chapter.html#scatter-gather。还有一点group-timeout 无论如何也帮不了你：不能保证一个组会及时过期。您需要使用表达式建议以任何方式将错误路由到聚合器。然后在发布函数中可以过滤最终列表中的错误：https://docs.spring.io/spring-integration/docs/5.0.8.RELEASE/reference/html/messaging-endpoints-chapter.html#message-handler-advice-chain