自定義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。

與此同時，您的解決方案是正確的解決方法。