自定义Resequencer行为

Question

我有一个场景，我必须发送消息为了休息服务，我打算使用resequencer。这再顺的行为必须是：

• 订单信息通过时间在白天（HH：MM：SS）：数据的消息仅

• 发布的消息后，便留了一段时间的总线（PE 2分钟）

在默认再顺并没有服务于这个目的，我决定开发一个自定义改变ResequencerMessageGroupProcessor的CustomResequencerMessageGroupProcessor。

我成功地使用了服务激活器，但我不得不明确地将输出通道定义为属性。没有办法在xml声明中使用output-channel属性？

当我使用输出信道属性出现以下错误：

Caused by: java.lang.IllegalArgumentException: no outputChannel or replyChannel header available 
at org.springframework.util.Assert.notNull(Assert.java:112) 
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.sendReplies(AbstractCorrelatingMessageHandler.java:616) 
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.completeGroup(AbstractCorrelatingMessageHandler.java:597) 
at org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler.handleMessageInternal(AbstractCorrelatingMessageHandler.java:405) 
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78) 
... 46 more 

这里是我的例子：

<int:channel id="resequencerChannel"/> 


<int:service-activator id="customResequencer" ref="resequencingMessageHandler" 
     input-channel="resequencerChannel" /> 

<int:channel id="aggregatedMessageChannel" /> 

<bean id="resequencingMessageHandler" class="org.springframework.integration.aggregator.ResequencingMessageHandler"> 
    <constructor-arg name="releaseStrategy" ref="timeoutReleaseStrategy"/> 
    <constructor-arg name="processor" ref="customResequencerMessageGroupProcessor"/> 
    <constructor-arg name="store" ref="redisMessageStore"/> 
    <constructor-arg name="correlationStrategy" ref="customCorrelationStrategy"/> 
    <property name="outputChannel" ref="aggregatedMessageChannel"/> 
    <property name="sendPartialResultOnExpiry" value="true"></property> 
</bean> 

<bean id="customResequencerMessageGroupProcessor" class="test.resequencer.CustomResequencerMessageGroupProcessor"> 
    <constructor-arg name="timeout" value="10000"/> 
</bean> 

<bean id="timeoutReleaseStrategy" class="org.springframework.integration.aggregator.TimeoutCountSequenceSizeReleaseStrategy" > 
    <constructor-arg name="threshold" value="100000"></constructor-arg> 
    <constructor-arg name="timeout" value="10000"/> 
</bean> 

<bean id="customCorrelationStrategy" class="org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy" > 
    <constructor-arg name="attributeName" value="correlationId"/> 

另外，如果你想有更好的方式来做到这一点，请，我会apreciatte告诉所以

在此先感谢！

问候

Guzman的

Answer 1

当引用（ref）一个从一个<service-activator/>的XML output-channel如果所引用的处理程序是一个AbstractReplyProducingMessageHandler（ARPMH）仅适用MessageHandler。

路由器，聚合器，再测试器等组件不被认为是ARPMH，因为它们有时会产生回复，有时候不会，并且在路由器的情况下可能会产生多个不适合的“回复”服务激活模型。

我们可以将聚合器/重定序器重构为AR​​PMH，因为它们只产生0或1个“回复”。如果参考是AbstractCorrelatingMessageHandler，我们还可以在ServiceActivatorFactoryBean中添加一些智能以注入输出通道。随意打开Improvement JIRA Issue。

与此同时，您的解决方案是正确的解决方法。