如何在 mule 中聚合入站 JMS 消息答案

【问题标题】：How to aggregate inbound JMS messages in mule如何在 mule 中聚合入站 JMS 消息
【发布时间】：2014-11-16 01:16:05
【问题描述】：

我从 activeMQ 接收到两条 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"/>
    <jms:outbound-endpoint queue="ReadOrder1" 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"/>
    <jms:outbound-endpoint queue="ReadOrder1" connector-ref="Active_MQ" doc:name="JMS">
        <jms:object-to-jmsmessage-transformer doc:name="Object to JMSMessage"/>
    </jms:outbound-endpoint>
</flow>

    <flow name="integration-flow" doc:name="integration-Flow1" processingStrategy="synchronous">
    <jms:inbound-endpoint queue="ReadOrder1" connector-ref="Active_MQ" doc:name="JMS"/>
    <vm:outbound-endpoint exchange-pattern="request-response" path="vm" doc:name="VM"/>
    <logger message="ending paylaod = #[payload]" level="INFO" doc:name="Logger"/>
</flow>
<flow name="integration-flow2" doc:name="integration-Flow2">
    <vm:inbound-endpoint exchange-pattern="request-response" path="vm" doc:name="VM"/>
    <collection-aggregator failOnTimeout="true" doc:name="Collection Aggregator"/>
</flow>

如何聚合传入的消息？我正在使用集合聚合器，并且不断收到消息 = 未设置相关组大小，但正在使用相关聚合器。消息按原样转发

消息是 JSON [{"salesOrderId":"00001-2-3","saleName":"Car Sale","status":"processing"}]

【问题讨论】：

我只能在流程中看到一个 'jms:inbound-endpoint' .. 你的第二个 JMS 在哪里？？
您好，谢谢您的回复，我已经添加了jms outbound
但是你说你需要聚合 2 条 jms 消息.. 在这里我只能看到来自 queue="ReadOrder1" 的 1 条 jms 消息
啊对不起，我明白你的意思了，我已经更新了流程

标签： jms mule activemq

【解决方案1】：

我在流程中看到了一些问题。在流程 clientoneFlow1 和 clienttwoFlow1 中，您都在使用 http 入站端点并分别点击 url 来启动流程调度消息到 queue="ReadOrder1" ..

但是，一旦消息被其中一个流传递到 queue="ReadOrder1"，integration-flow 就会开始，并且消息会再次发送到 integration-flow2 由 collection-aggregator 接收，它不会等待其他流获取消息（因为您需要点击其他流的 url 来启动它）。 ..

所以.. 在这里您可以让流 clientoneFlow1 和 clienttwoFlow1 几乎并行执行并将消息发送到 collection-aggregator 几乎同时..

所以.. 一种方法可以通过使用分散和收集组件来实现它，并且您不需要 collection-aggregator 那里..

来自所有流的所有消息将在流结束时自动聚合..

例如...在这里您可以执行以下操作：-

  <flow name="fork" doc:name="fork">
   <http:inbound-endpoint host="localhost" port="8090" path="scattergather" exchange-pattern="request-response" doc:name="HTTP"/>

  <scatter-gather timeout="6000">
   <!-- Calling clientoneFlow1-->
   <flow-ref name="clientoneFlow1" doc:name="Flow Reference"/>

   <!-- Calling clienttwoFlow1-->
   <flow-ref name="clienttwoFlow1" doc:name="Flow Reference"/>
   </scatter-gather>

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

 <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"/>

 </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"/>
 </flow>

现在这是从两个流中实现组合有效负载的最简单方法，而不会使它变得如此复杂..

【讨论】：