我有一个简单的驼峰路由,它接受一个项目列表,将它们分开,将每个元素发送到mq节点进行处理,然后通过聚合器将它们连接回一起。
非常接近组合消息处理器:http://camel.apache.org/composed-message-processor.html
但是我们注意到,在分裂之后,camel将创建多个并发消费者?还是交流?由于消息被发送给多个消费者,因此它们永远不会完成。
清单:1、2、3、4
分裂:amq:: process_each_item
总:[Camel (camel-3) thread #41 - Aggregating 1 - Waiting on 3 more items
[Camel (camel-1) thread #16 - Aggregating 2 - Waiting on 3 more items
[Camel (camel-3) thread #49 - Aggregating 3 - Waiting on 2 more items
[Camel (camel-1) thread #15 - Aggregating 4 - Waiting on 2 more items
因此,camel产生了2个聚合器,每个聚合器等待4个条目,但它们每个只获得两个。
骆驼路线:
<camelContext xmlns="http://camel.apache.org/schema/spring">
<route> <!-- This route splits the reg request into it's items. Adding needed info to the message header. -->
<from uri="activemq:registration.splitByItemQueue" /> <!-- pick up the reg req -->
<setHeader headerName="regReqId"> <!-- Need to store the Reg Req in the header -->
<simple>${body.registrationRequest.id}</simple>
</setHeader>
<split parallelProcessing="false" strategyRef="groupedExchangeAggregator"> <!-- Split the RegRequestInfo into it's individual requestItems (add, drop, etc) -->
<method ref="requestSplitter" method="split" /> <!-- does the actual splitting -->
<setHeader headerName="JMSXGroupID"> <!-- This is CRITICAL. It is how we ensure valid seat check counts without db locking -->
<simple>FOID=${body.formatOfferingId}</simple> <!-- grouping on the foid -->
</setHeader>
<to uri="activemq:registration.lprActionQueue"/> <!-- send to queue's for processing-->
</split>
</route>
<route> <!-- performs the registration + seat check -->
<from uri="activemq:registration.lprActionQueue" />
<bean ref="actionProcessor" method="process"/> <!-- go to the java code that makes all the decisions -->
<to uri="activemq:registration.regReqItemJoinQueue"/> <!-- send to join queue's for final processing-->
</route>
<route> <!-- This route joins items from the reg req item split. Once all items have completed, update state-->
<from uri="activemq:registration.regReqItemJoinQueue" /> <!-- Every Reg Req Item will come here-->
<aggregate strategyRef="groupedExchangeAggregator" ignoreInvalidCorrelationKeys="false" completionFromBatchConsumer="true"> <!-- take all the Reg Req Items an join them to their req -->
<correlationExpression>
<header>regReqId</header> <!-- correlate on the regReqId we stored in the header -->
</correlationExpression>
<bean ref="actionProcessor" method="updateRegistrationRequestStatus"/> <!-- update status -->
</aggregate>
</route>
</camelContext>
<bean id="groupedExchangeAggregator" class="org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy" />
在我的本地机器上,上面的工作正常,但是当我们部署到我们的测试服务器时,一半的消息到一个驼峰聚合器,一半到另一个。导致没有人完成。注意,在下面的配置中,我们将camel的并发消费者设置为1。
这是camel/activemq配置
<amq:broker useJmx="false" persistent="false">
<amq:plugins>
<amq:statisticsBrokerPlugin />
</amq:plugins>
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:0" />
</amq:transportConnectors>
</amq:broker>
<!-- Basic AMQ connection factory -->
<amq:connectionFactory id="amqConnectionFactory" brokerURL="vm://localhost" />
<!-- Wraps the AMQ connection factory in Spring's caching (ie: pooled) factory
From the AMQ "Spring Support"-page: "You can use the PooledConnectionFactory for efficient pooling... or you
can use the Spring JMS CachingConnectionFactory to achieve the same effect."
See "Consuming JMS from inside Spring" at http://activemq.apache.org/spring-support.html
Also see http://codedependents.com/2010/07/14/connectionfactories-and-caching-with-spring-and-activemq/
Note: there are pros/cons to using Spring's caching factory vs Apache's PooledConnectionFactory; but, until
we have more explicit reasons to favor one over the other, Spring's is less tightly-coupled to a specific
AMQP-implementation.
See http://stackoverflow.com/a/19594974
-->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="amqConnectionFactory"/>
<property name="sessionCacheSize" value="1"/>
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<constructor-arg ref="connectionFactory" />
</bean>
<bean id="jmsConfig"
class="org.apache.camel.component.jms.JmsConfiguration">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="concurrentConsumers" value="1"/>
</bean>
<bean id="activemq"
class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="configuration" ref="jmsConfig"/>
</bean>
原来我们有另一个spring上下文/servlet导入我们的配置。我们认为这就是问题所在。