JMS 从同步到异步 [英] JMS from Synchronous to Asynchronous
问题描述
我能够使用 JMS 创建同步服务总线，但我无法将其转换为异步.我正在尝试向服务异步发送请求，因此如果服务关闭，我希望 JMS 队列保留请求消息，并在服务启动时将消息传递给服务并返回响应.
这是我的代码
<路线><from id ="server" uri="jetty:http://0.0.0.0:9500/rsb/toService?matchOnUriPrefix=true&enableMultipartFilter=false&disableStreamCache=false"/><wireTap uri="log:test?level=INFO"><body><simple>进入JMS测试路由并发送消息到队列</simple></body></wireTap><!--<to uri="direct:queue"/>--><to uri="jms://testqueue?requestTimeout=360000&replyTo=bar&replyToDeliveryPersistent=true&exchangePattern=InOut&acknowledgementModeName=AUTO_ACKNOWLEDGE"/></路线><route id="testqueuelistener" streamCache="true"><from uri="jms://testqueue?replyToDeliveryPersistent=true&replyTo=bar"/><wireTap uri="log:test?level=INFO"><body><simple>从队列收到的消息:${body}</simple></body></wireTap><to uri="http://localhost:18402/Home/addUser?bridgeEndpoint=true&throwExceptionOnFailure=false"/><to uri="jms:service"/></路线><路线><from uri="jms:service"/><变换><simple>${body}</simple></转换></路线></camelContext> 问题是您在不使用事务的情况下访问 JMS 队列 - 因此，一旦您收到消息，它就会从队列中消失.您需要使用事务，并且只有在完成处理后才提交(或回滚)消息消费.
相关的企业集成模式是事务客户端.JMS 组件文档 还提供了一些有关事务的信息.最后，Camel in Action 的第 9 章(第二版的第 12 章)专门介绍交易(我怎么推荐都不为过！).
您需要:
- 获取 JMS 事务管理器(您使用哪个事务管理器可能取决于您的特定用例)
- 配置 Camel JMS 组件以使用事务管理器
- 使用事务策略来配置路由的事务行为(或者只是将路由标记为事务，并使用默认策略)
配置看起来像:
<osgi:reference id="jmsConnectionPool" interface="javax.jms.ConnectionFactory"/><!-- 我们创建了一个 Spring JmsTransactionManager(我们的事务管理器也可以是一个导入 OSGi 服务，就像我们为连接工厂所做的那样；例如一个 XA 事务经理)--><bean id="jmsTxManager" class="org.springframework.jms.connection.JmsTransactionManager"><property name="connectionFactory" ref="jmsConnectionPool"/></bean><!-- 我们将 JMS 组件配置为使用事务管理器--><bean id="jms" class="org.apache.camel.component.jms.JmsComponent"><property name="connectionFactory" ref="jmsConnectionPool"/><property name="transacted" value="true"/><property name="transactionManager" ref="jmsTxManager"/></bean><!-- 这是一个交易策略的例子--><bean id="requiresNew" class="org.apache.camel.spring.spi.SpringTransactionPolicy"><property name="transactionManager" ref="jtaTransactionManager"/><property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW"/></bean>这是一条交易路线:
<from uri="jms://..."/><交易/>...</路线> 如果需要，路由可以使用特定的事务策略:
<from uri="jms://..."/><transacted ref="requiresNew"/>...</路线> I was able to create a Synchronous service bus using the JMS, but I was not able to turn it to Asynchronous. I'm trying to post a request to a service asynchronously, so if the service is down, I want the JMS queue keeps the request message and when the service starts up it delivers the message to the service and get back the response.
here is my code
<camelContext xmlns="http://camel.apache.org/schema/spring">
<route >
<from id ="server" uri="jetty:http://0.0.0.0:9500/rsb/toService?matchOnUriPrefix=true&enableMultipartFilter=false&disableStreamCache=false"/>
<wireTap uri="log:test?level=INFO"><body><simple>Enter JMS test route and sending message to queue</simple></body></wireTap>
<!--<to uri="direct:queue"/>-->
<to uri="jms://testqueue?requestTimeout=360000&replyTo=bar&replyToDeliveryPersistent=true&exchangePattern=InOut&acknowledgementModeName=AUTO_ACKNOWLEDGE"/>
</route>
<route id="testqueuelistener" streamCache="true">
<from uri="jms://testqueue?replyToDeliveryPersistent=true&replyTo=bar" />
<wireTap uri="log:test?level=INFO"><body><simple>Message recieved from queue: ${body}</simple></body></wireTap>
<to uri="http://localhost:18402/Home/addUser?bridgeEndpoint=true&throwExceptionOnFailure=false"/>
<to uri="jms:service"/>
</route>
<route >
<from uri="jms:service"/>
<transform>
<simple>${body}</simple>
</transform>
</route>
</camelContext>
The issue is that you are accessing the JMS queue without using a transaction - so as soon as you get the message, it's gone from the queue. You need to use a transaction and only commit (or rollback) the message consumption after you've finished processing it.
The relevant Enterprise Integration Pattern is the transactional client. The JMS component documentation also provides some information about transaction. Finally, chapter 9 of Camel in Action (chapter 12 for the second edition) is dedicated to transactions (and I can't recommend it enough!).
You need to:
- Obtain a JMS transaction manager (which transaction manager you use may depend on your specific use case)
- Configure the Camel JMS component to use the transaction manager
- Use a transaction policy to configure the transactional behavior of your route (or just mark the route as transacted, and use the default policy)
The configuration can look something like:
<!-- Import JMS connection factory -->
<osgi:reference id="jmsConnectionPool" interface="javax.jms.ConnectionFactory" />
<!-- We create a Spring JmsTransactionManager (our transaction manager could also be an
imported OSGi service, like we do for the connection factory; for example an XA transaction
manager) -->
<bean id="jmsTxManager" class="org.springframework.jms.connection.JmsTransactionManager">
<property name="connectionFactory" ref="jmsConnectionPool"/>
</bean>
<!-- We configure the JMS component to use the transaction manager-->
<bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
<property name="connectionFactory" ref="jmsConnectionPool" />
<property name="transacted" value="true"/>
<property name="transactionManager" ref="jmsTxManager"/>
</bean>
<!-- Here's an example of a transaction policy -->
<bean id="requiresNew" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
<property name="transactionManager" ref="jtaTransactionManager"/>
<property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW"/>
</bean>
Here's a transacted route:
<route id="myRoute">
<from uri="jms://..." />
<transacted/>
...
</route>
And a route can use a specific transaction policy if we want:
<route id="myRoute">
<from uri="jms://..." />
<transacted ref="requiresNew" />
...
</route>
这篇关于JMS 从同步到异步的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
