在mule中聚合后获取JMS Correlation Id答案

【问题标题】：Getting JMS Correlation Id after aggregation in mule在mule中聚合后获取JMS Correlation Id
【发布时间】：2015-01-13 12:15:45
【问题描述】：

我有两条 JMS 出站消息，它们在分散聚集中接收，用于聚合来自客户端的两条消息。然后我需要使用拆分器拆分消息，然后通过 jms 将它们一个一个发送到另一个流：

<flow name="clientoneFlow1" doc:name="clientoneFlow1">
    <http:inbound-endpoint exchange-pattern="one-way" host="localhost" port="8081" path="client1" doc:name="HTTP"/>
    <component class="SalesOrder" doc:name="Java"/>
    <json:object-to-json-transformer doc:name="Object to JSON"/>
    <set-property propertyName="MULE_CORRELATION_ID" value="clientOne" doc:name="Property"/>
    <set-property propertyName="MULE_CORRELATION_SEQUENCE" value="1" doc:name="Property"/>
    <jms:outbound-endpoint queue="client1.publish" connector-ref="Active_MQ" doc:name="JMS">
        <jms:object-to-jmsmessage-transformer doc:name="Object to JMSMessage"/>
    </jms:outbound-endpoint>
</flow>

<flow name="clienttwoFlow1" doc:name="clienttwoFlow1">
        <http:inbound-endpoint exchange-pattern="one-way" host="localhost" port="8081" path="client2" doc:name="HTTP"/>
    <component class="SalesOrder2" doc:name="Java"/>
    <json:object-to-json-transformer doc:name="Object to JSON"/>
    <set-property propertyName="MULE_CORRELATION_ID" value="clientTwo" doc:name="Property"/>
    <set-property propertyName="MULE_CORRELATION_SEQUENCE" value="2" doc:name="Property"/>
    <jms:outbound-endpoint queue="client2.publish" connector-ref="Active_MQ" doc:name="JMS">
        <jms:object-to-jmsmessage-transformer doc:name="Object to JMSMessage"/>
    </jms:outbound-endpoint>
</flow>

<flow name="integration-internetsolutionsFlow1" doc:name="integration-internetsolutionsFlow1">
    <http:inbound-endpoint exchange-pattern="request-response" host="localhost" port="8081" doc:name="HTTP" path="esb"/>
    <scatter-gather doc:name="Scatter-Gather">
        <processor-chain>
            <flow-ref name="clientoneFlow1" doc:name="Flow Reference"/>
            <logger message="Client One: #[message.correlationSequence]" level="INFO" doc:name="Logger"/>
        </processor-chain>
        <processor-chain>
            <flow-ref name="clienttwoFlow1" doc:name="Flow Reference"/>
            <logger message="Client Two: #[message.correlationSequence]" level="INFO" doc:name="Logger"/>
        </processor-chain>
    </scatter-gather>
    <logger level="INFO"  message="Combined Payload: #[message.payload]" doc:name="Logger"/>
    <splitter enableCorrelation="ALWAYS" expression="#[payload]" doc:name="Splitter"/>
    <logger message="after splitter = #[payload]" level="INFO" doc:name="Logger"/>
    <logger message="Corr after splitter = #[message.correlationId] and group = #[message.correlationGroupSize]" level="INFO" doc:name="Logger"/>
    <jms:outbound-endpoint queue="validation.queue" doc:name="JMS"/>
</flow>

<flow name="validateFlow1" doc:name="validateFlow1">
    <jms:inbound-endpoint doc:name="JMS" connector-ref="Active_MQ" queue="validation.queue"/>
    <logger message="splinter payload = #[payload]" level="INFO" doc:name="Logger"/>
    <logger message="splitter corr Id = #[message.correlationId]" level="INFO" doc:name="Logger"/>
</flow>

但是在拆分器之后，两个客户端的关联 Id 已经消失并分配了相同的关联 Id。如何在拆分器后检索相关 ID？并在验证流程中根据correlationId消费消息

【问题讨论】：

为什么要使用拆分器？？
感谢您的回复，我需要拆分消息，编辑单个消息中的数据，编辑完数据后再次聚合数据。然后发回消息。

标签： jms mule

【解决方案1】：

如果您使用表达式拆分器，您将使用有效负载列表进行拆分，因此每个拆分的分支上都有一个新的 MuleEvent。

如果您使用集合拆分器，您将使用EventToMessageSequenceSplittingStrategy，它将正确处理MessageCollection 的拆分。

因此，将 splitter 替换为 collection-splitter。

【讨论】：

【解决方案2】：

在 scatter-gather ....

之后，您不需要拆分器来拆分消息

在scatter-gather之后，消息会自动合并，并且可以按以下形式提取消息：- #[message.payload[0]] 和 #[message.payload[1]] 示例：-

<logger level="INFO"  message="Payload1 of clientoneFlow1 : #[message.payload[0]] and clienttwoFlow1: #[message.payload[1]] " doc:name="Logger"/>

所以你可以删除分离器并尝试

【讨论】：