如何在骡子消息上设置相关序列属性

Question

我想在使用集合聚合器进行聚合之前对来自jdbc查询的消息进行排序，我认为这样做的最佳方法是resequencer。当我添加这行

<resequencer timeout="6000" failOnTimeout="false" />

工作流产生如下错误:

* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * 消息 : 表达式的执行"message.payload。将INVOICE_NUMBER"失败了。(org.mule.api.expression.ExpressionRuntimeException)。消息有效负载是类型:MuleEvent[] Code: MULE_ERROR——2-------------------------------------------------------------------------------- 异常堆栈是:1. [Lorg.mule.api.MuleEvent;不能强制转换为java.util.Map (java.lang.ClassCastException)
org.mvel2.optimizers.impl.refl.nodes。MapAccessor: 42(空)2. [错误:无法调用getter: getPayload] declr.class: org.mule.el.context.MessageContext;act.class:org.mule.el.context。(参见trace)] [Near:{…未知的…}]^ [Line: 1, Column: 0] (org.mvel2.CompileException)GetterAccessor: 75(空)3.表达式"message.payload"的执行。将INVOICE_NUMBER"失败了。(org.mule.api.expression.ExpressionRuntimeException)
org.mule.el.mvel.MVELExpressionLanguage: 211(http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/api/expression/ExpressionRuntimeException.html)4. 表达式"message.payload"的执行。将INVOICE_NUMBER"失败了。(org.mule.api.expression.ExpressionRuntimeException)。消息负载类型为:MuleEvent[](org.mule.api.transformer.TransformerMessagingException)
org.mule.transformer.AbstractTransformer: 123(http://www.mulesoft.org/docs/site/current3/apidocs/org/mule/api/transformer/TransformerMessagingException.html)-------------------------------------------------------------------------------- 根异常堆栈跟踪:java.lang.ClassCastException:[Lorg.mule.api.MuleEvent;不能强制转换为java.util. mapatorg.mvel2.optimizers.impl.refl.nodes.MapAccessor.getValue (MapAccessor.java: 42)org.mvel2.optimizers.impl.refl.nodes.GetterAccessor.getValue (GetterAccessor.java: 41)在org.mvel2.optimizers.impl.refl.nodes.VariableAccessor.getValue (VariableAccessor.java: 38)+ 3 more(设置调试级别日志记录或'-Dmule.verbose ')。Exceptions =true' for everything)

---

<when expression="#[message.payload.LHF_INVOICE_METHOD == 'EDI']"><processor-chain>
        <message-properties-transformer doc:name="Set Correlation ID">
            <add-message-property key="MULE_CORRELATION_SEQUENCE" value="#[message.payload.INVOICE_NUMBER]" />
        </message-properties-transformer>
<resequencer timeout="6000" failOnTimeout="false" />
<set-payload value="#[message.payload.INVOICE_NUMBER]" doc:name="EDI"/>                                
<collection-aggregator  timeout="3200" failOnTimeout="false" doc:name="EDI"          storePrefix="EDI" />
<smtp:outbound-endpoint host="mail.example.com" to="[e-mail]" from="[e-mail]" subject="[Invoice Workflow] EDI" responseTimeout="10000" doc:name="EDI"/>
<collection-splitter doc:name="Collection Splitter"/>
<jdbc:outbound-endpoint exchange-pattern="one-way" queryKey="gMarkProcessedInRoss"     queryTimeout="1000" connector-ref="DatabaseMuleLogin" doc:name="Mark Processed in Ross">
</jdbc:outbound-endpoint>
</processor-chain>
</when>

我在resequencer(下图)之后添加了一个分离器，它可以工作，除了消息仍然在电子邮件中未排序。

<set-property propertyName="MULE_CORRELATION_SEQUENCE" value="#[message.payload.INVOICE_NUMBER]" />
<resequencer timeout="6000" failOnTimeout="false" />
<collection-splitter doc:name="Collection Splitter"/>
<set-payload value="#[message.payload.message.payload.INVOICE_NUMBER]" doc:name="EDI"/>
<echo-component doc:name="Echo"/>                                
<collection-aggregator  timeout="3200" failOnTimeout="false" doc:name="EDI"  storePrefix="EDI"       />
<smtp:outbound-endpoint host="mail.example.com" to="[e-mail]" from="[e-mail]" subject="[Invoice Workflow] EDI" responseTimeout="10000" doc:name="EDI"/>
<collection-splitter doc:name="Collection Splitter"/>
<jdbc:outbound-endpoint exchange-pattern="one-way" queryKey="RouteEDI" queryTimeout="1000" connector-ref="DatabaseMuleLogin" doc:name="Mark Processed in Ross">
</jdbc:outbound-endpoint>

我添加了一个echo组件，它们在那个点上是排序的，所以我很确定聚合器正在取消排序。

Answer 1

您是正确的，相关序列属性命名为MULE_CORRELATION_SEQUENCE。您可以如下所示进行设置:

<set-property propertyName="MULE_CORRELATION_SEQUENCE"
              value="1" />

然后，resequencer应该根据该序列对消息进行排序。如果没有，请用失败的配置更新您的问题。

确保将MULE_CORRELATION_GROUP_SIZE属性设置为正确的大小，并且所有要重新排序的消息共享相同的MULE_CORRELATION_ID属性。