本文简单介绍一下Apache Camel如何对route进行事务控制,首先介绍整个route只涉及到一个事务参与者的情况,然后再介绍route中涉及到多个事务参与者的情况.Camel是通过和Spring的框架集成进行事务控制的.
1,整个route只有一个事务参与者,"局部事务",这里用JMS的例子,后台的MQ为ActiveMQ,示例图如下:
route的代码如下:
public class JMSTransaction extends RouteBuilder {
public void configure() throws Exception {
TProcessor0 p0 = new TProcessor0();
TProcessor1 p1 = new TProcessor1();
from("jms:queue:TOOL.DEFAULT").process(p0).process(p1).to("file:d:/temp/outbox");
}
}
Spring配置如下:
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:broker="http://activemq.apache.org/schema/core"
xsi:schemaLocation="
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://camel.apache.org/schema/spring/camel-spring.xsd
http://activemq.apache.org/schema/core/activemq-core-5.2.0.xsd">
<camelContext xmlns="http://camel.apache.org/schema/spring">
<package>
com.test.camel.transaction.jms
</package>
</camelContext>
<bean>
<property value="true"/>
<property ref="txManager"/>
</bean>
<bean>
<property ref="jmsConnectionFactory"/>
</bean>
<bean>
<property value="tcp://localhost:61616"/>
</bean>
</beans>
route定义的逻辑为从queue里取消息,然后进行一系列的处理(process(p0).process(p1)),<property value="true"/>的意思是通过这个jms进行的消息存取是有事务控制的.上面的route在process(p1)里抛出异常,txManager会进行rollback处理.(在activeMQ里,消息默认会redelivery到客户端6次,如果继续异常,消息会放到deadletter queue里(ActiveMQ.DLQ)),需要在AciveMQ的配置文件activemq.xml里配置如下:(non-persistent的queue的消息出错也转到dead letter queue)
<policyEntry queue=">">
<deadLetterStrategy>
<sharedDeadLetterStrategy processNonPersistent="true" />
</deadLetterStrategy>
如果<property value="false"/>的话,消息在重发了6次后会丢失.
如果上面例子中的事务参与者是数据库的话,道理与之类似,只是配置的transaction manager不同,如:
<bean/>