Apache Camel - 事务

水深无声 2022-02-16 01:21 247阅读 0赞

> Apache Camel是一个基于Enterprise Integration Pattern(企业整合模式，简称EIP)的开源框架。

### 1. 概述 ###

关于`Apache Camel`的介绍就不说了，作为2007年就发布了v1.0版的开源项目，至今拥有着非常强大的生命力（Github上几乎每天都会有更新），相关的介绍在网上随处可见，本文主要关注的是Camel在事务方面的支持。

### 2. 本地事务和全局事务 ###

###### 2.1 本地事务 ######

本地事务又称单事务，这也是我们最常接触到的事务类型。Camel也是直接选择了复用Spring的事务支持，诚如《人月神话》第四章里提到的——法国城市南斯的建筑历史上，八代建筑者自我约束和牺牲精神，确保了最终建筑风格的一致性。针对事务支持，Camel成功遏制了自创一套实现的冲动，转而寻求既有稳定的实现方式，这一点上真的非常难能可贵。

以下我们使用Camel-Sql来给出一个例子：

============================ camel-spring.xml 文件

<context:property-placeholder location="classpath:db.properties" />
    
    <bean id="dataSource" class="org.springframework.jdbc.datasource.TransactionAwareDataSourceProxy">
    	<constructor-arg>
    		<bean class="com.alibaba.druid.pool.DruidDataSource">
    			<property name="driverClassName" value="${jdbc.driverClassName}" />
    			<property name="url" value="${jdbc.url}" />
    			<property name="username" value="${jdbc.username}" />
    			<property name="password" value="${jdbc.password}" />
    		</bean>
    	</constructor-arg>
    </bean>
    
    <bean id="txManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    	<property name="dataSource" ref="dataSource"/>
    </bean>	
    
    
    
    
    <bean id="camelTransaction" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    	<property name="transactionManager" ref="txManager"/>
    	<property name="propagationBehaviorName" value="PROPAGATION_REQUIRED"/>
    </bean>
    
    
    <bean id="sqlComponent" class="org.apache.camel.component.sql.SqlComponent">
    	<property name="dataSource" ref="dataSource" />
    </bean>
    
    
    <camelContext xmlns="http://camel.apache.org/schema/spring">
    
    	
    	<propertyPlaceholder id="placeholder" location="classpath:sql/sql.properties" />
    
    	
    	<route id="generateOrder-route">
    		
    		<from uri="stream:in?promptMessage=Enter something:" />
    		
    		<transacted ref="camelTransaction"/>
    		<to uri="sqlComponent:{ {sql.insertNewTopic}}" />
    		
    		<throwException exceptionType="java.lang.RuntimeException" message="模拟异常, 将导致数据库事务回滚!"/>
    	</route>
    </camelContext>

============================ sql.properties 文件

# sql.properties 文件
    sql.insertNewTopic=INSERT INTO topic(id, hh) VALUES (1, 'topic')

============================ 单元测试代码

@Test
    public void singleTransaction() throws Exception { 
    	ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
    			"camel-spring.xml", this.getClass());
    	context.start();
    	
    	// 通用没有具体业务意义的代码，只是为了保证主线程不退出
    	Object obj = new Object();
    	synchronized (obj.getClass()) { 
    		obj.getClass().wait();
    	}
    	
    }

###### 2.2 全局事务 ######

全局事务又称分布式事务，这里就需要使用诸如Atomikos这样的第三方组件的支持了，在实际的项目中，使用这样的方式实现分布式事务对性能影响较大，导致应用场景有限，所以这里就不给出相关的示例代码了。

###### 2.3 提示 ######

Camel提供的数据库相关组件中，camel-sql, camel-mybatis均支持事务，但camel-jdbc是不支持事务的， 这一点要注意。可见参见链接：[关于camel-jdbc支持事务的讨论(2011年)][camel-jdbc_2011]

### 3. 技巧 ###

###### 3.1 事务失败的时候返回自定义值 ######

我们偶尔会遇到如小节标题所示的需求，针对这种情况，Camel也早已给出了相应的通用解决方案 —— 使用 `onException` ：

<onException>
    	
    	<exception>java.lang.RuntimeException</exception>
    	
    	<handled><constant>true</constant></handled>
    	
    	<transform>constant>fail</constant></transform>
    	
    	<rollback markRollbackOnly="true"/>
    </onException>

###### 3.2 事务补偿 ######

事务一般都是存在于数据库，MQ之间，诸如文件操作等其实是不存在所谓事务操作的，针对此类情况Camel也是给出了相应的解决方案：

**1. `UnitOfWork`和`Synchronization`**  
Camel通过引入`UnitOfWork`来模拟事务。

CamelTestUtil.defaultPrepareTest(new RouteBuilder() { 
    	@Override
    	public void configure() throws Exception { 
    		from("stream:in?promptMessage=Enter something:")// 
    				.process(new Processor() { 
    					public void process(Exchange exchange) throws Exception { 
    					   // MyRollback实现了Synchronization接口的onFailure()方法, 将在执行失败时候回调, 这也就是事务补偿的时机
    						exchange.getUnitOfWork().addSynchronization(new MyRollback());
    					}
    				})//
    				......					
    				.throwException(new RuntimeException("XXXXXX"))// 
    				.to("stream:out");
    	}
    });

**2. `onCompletion`**  
实现事务补偿的另外一种方式就是使用Camel提供的`onCompletion`特性，Camel保证其在路由结束时候被调用。你可已通过在其后附加`onFailureOnly()`来选择只有在失败时候才调用。

from("stream:in?promptMessage=Enter something:")// 
    	// use a route scoped onCompletion to be executed when the Exchange failed
    	.onCompletion().onFailureOnly().bean(MyRollback.class, "onFailure")//
    	// must use end() to denote the end of the onCompletion route! 这里切记, 笔者就栽了跟头.
    	.end() // 
    	// here starts the regular route
    	.process(ThrowExceptionXProcessor.me)//
    	.to("stream:out");

**3. 以上两种方法的区别**  
两者最大的区别是其线程模型：

1.  `Synchronization`，所有的操作都在同一线程下
2.  `onCompletion`，新起单独的线程专门处理，所以`onCompletion`中对于Exchange的修改等等不会影响主线程中的，`onCompletion`获取到的Exchange属于复制品。

### 4. 最后 ###

1.  社区强大真是一件无比幸福的事情。Apache Camel在国内的资料并不多，但依托于强大的生态圈，你依然可以相当轻松地渡过初期阶段。
2.  《Camel In Action》这本书相当不错，面面俱到，关于使用Camel的问题，你已经遇到的或者之后会遇到的问题，他都给出了相应的讨论。

### 5. 参考 ###

1.  《Camel In Action》 P283 - P314
2.  [Camel - Transaction][]

[camel-jdbc_2011]: https://issues.apache.org/jira/browse/CAMEL-3803
[Camel - Transaction]: http://people.apache.org/~dkulp/camel/transactional-client.html