Spring中使用atomikos+druid实现经典分布式事务

Dear 丶 2022-10-02 12:45 224阅读 0赞

经典分布式事务，是相对互联网中的柔性分布式事务而言，其特性为ACID原则，包括原子性（Atomictiy）、一致性（Consistency）、隔离性（Isolation）、持久性（Durabilit）：

*  原子性：事务是一个包含一系列操作的原子操作。事务的原子性确保这些操作全部完成或者全部失败。
 *  一致性：一旦事务的所有操作结束，事务就被提交。然后你的数据和资源将处于遵循业务规则的一直状态。
 *  隔离性：因为同时在相同数据集上可能有许多事务处理，每个事务应该与其他事务隔离，避免数据破坏。
 *  持久性：一旦事务完成，他的结果应该能够承受任何系统错误（想象一下在事务提交过程中机器的电源被切断的情况）。通常，事务的结果被写入持续性存储。

XA是啥？  
XA是由X/Open组织提出的分布式事务的架构（或者叫协议）。XA架构主要定义了（全局）事务管理器（Transaction Manager）和（局部）资源管理器（Resource Manager）之间的接口。XA接口是双向的系统接口，在事务管理器（Transaction Manager）以及一个或多个资源管理器（Resource Manager）之间形成通信桥梁。也就是说，在基于XA的一个事务中，我们可以针对多个资源进行事务管理，例如一个系统访问多个数据库，或即访问数据库、又访问像消息中间件这样的资源。这样我们就能够实现在多个数据库和消息中间件直接实现全部提交、或全部取消的事务。XA规范不是java的规范，而是一种通用的规范，  
目前各种数据库、以及很多消息中间件都支持XA规范。  
JTA是满足XA规范的、用于Java开发的规范。所以，当我们说，使用JTA实现分布式事务的时候，其实就是说，使用JTA规范，实现系统内多个数据库、消息中间件等资源的事务。

JTA(Java Transaction API)，是J2EE的编程接口规范，它是XA协议的JAVA实现。它主要定义了：

*  一个事务管理器的接口javax.transaction.TransactionManager，定义了有关事务的开始、提交、撤回等>操作。
 *  一个满足XA规范的资源定义接口javax.transaction.xa.XAResource，一种资源如果要支持JTA事务，就需要让它的资源实现该XAResource接口，并实现该接口定义的两阶段提交相关的接口。如果我们有一个应用，它使用JTA接口实现事务，应用在运行的时候，就需要一个实现JTA的容器，一般情况下，这是一个J2EE容器，像JBoss，Websphere等应用服务器。但是，也有一些独立的框架实现了JTA，例如**Atomikos, bitronix**都提供了jar包方式的JTA实现框架。这样我们就能够在Tomcat或者Jetty之类的服务器上运行使用JTA实现事务的应用系统。在上面的本地事务和外部事务的区别中说到，JTA事务是外部事务，可以用来实现对多个资源的事务性。它正是通过每个资源实现的XAResource来进行两阶段提交的控制。感兴趣的同学可以看看这个接口的方法，除了commit, rollback等方法以外，还有end(), forget(), isSameRM(), prepare()等等。光从这些接口就能够想象JTA在实现两阶段事务的复杂性。

本篇以Spring MVC+Maven+Atomikos+Druid+MyBatis演示分布式事务的实现。

**Mave 的pom.xml**

<properties>
    	<jdk.version>1.8</jdk.version>
    	
    	<mysql.version>8.0.11</mysql.version>
    	<druid.version>1.1.17</druid.version>
    	<spring.version>5.1.8.RELEASE</spring.version>
    	<cglib.version>3.2.12</cglib.version>
    	<atomikos.version>5.0.0</atomikos.version>
    	<aspectjweaver.version>1.9.4</aspectjweaver.version>
    	<aspectjrt.version>1.5.4</aspectjrt.version>
    	<jta.version>1.1</jta.version>
    	<mybatise.version>3.2.0</mybatise.version>
    	<mybatis.spring>1.2.0</mybatis.spring>
    	<log4j.version>1.2.17</log4j.version>
    	<junit.version>4.12</junit.version>
    	<cglib.version>3.2.4</cglib.version>
    </properties>
    <dependencies>
    	<dependency>
    		<groupId>org.mybatis</groupId>
    		<artifactId>mybatis</artifactId>
    		<version>${mybatise.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.mybatis</groupId>
    		<artifactId>mybatis-spring</artifactId>
    		<version>${mybatis.spring}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.atomikos</groupId>
    		<artifactId>atomikos-util</artifactId>
    		<version>${atomikos.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.atomikos</groupId>
    		<artifactId>transactions</artifactId>
    		<version>${atomikos.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.atomikos</groupId>
    		<artifactId>transactions-jta</artifactId>
    		<version>${atomikos.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.atomikos</groupId>
    		<artifactId>transactions-jdbc</artifactId>
    		<version>${atomikos.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.atomikos</groupId>
    		<artifactId>transactions-api</artifactId>
    		<version>${atomikos.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>javax.transaction</groupId>
    		<artifactId>jta</artifactId>
    		<version>${jta.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>cglib</groupId>
    		<artifactId>cglib-nodep</artifactId>
    		<version>${cglib.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-test</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-web</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-tx</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-beans</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-jdbc</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-webmvc</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-orm</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>org.springframework</groupId>
    		<artifactId>spring-context-support</artifactId>
    		<version>${spring.version}</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>org.aspectj</groupId>
    		<artifactId>aspectjweaver</artifactId>
    		<version>${aspectjweaver.version}</version>
    	</dependency>
    
    	<dependency>
    		<groupId>aspectj</groupId>
    		<artifactId>aspectjrt</artifactId>
    		<version>${aspectjrt.version}</version>
    	</dependency>
    	
    	<dependency>
    		<groupId>cglib</groupId>
    		<artifactId>cglib</artifactId>
    		<version>${cglib.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>mysql</groupId>
    		<artifactId>mysql-connector-java</artifactId>
    		<version>${mysql.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>com.alibaba</groupId>
    		<artifactId>druid</artifactId>
    		<version>${druid.version}</version>
    	</dependency>
    	<dependency>
    		<groupId>junit</groupId>
    		<artifactId>junit</artifactId>
    		<version>${junit.version}</version>
    		<scopte>test</scope>
    	</dependency>
    </dependencies>

**spring-application-context.xml**

**spring-mybatis-atomikos-druid.xml**

<?xml version="1.0" encoding="UTF-8"?>
    <beans  xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
        xmlns:jee="http://www.springframework.org/schema/jee"
        xmlns:tx="http://www.springframework.org/schema/tx" 
        xmlns:context="http://www.springframework.org/schema/context" 
        xmlns:aop="http://www.springframework.org/schema/aop"
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans-3.0.xsd 
        http://www.springframework.org/schema/tx 
        http://www.springframework.org/schema/tx/spring-tx-3.0.xsd 
        http://www.springframework.org/schema/jee 
        http://www.springframework.org/schema/jee/spring-jee-3.0.xsd 
        http://www.springframework.org/schema/aop 
        http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
        http://www.springframework.org/schema/context 
        http://www.springframework.org/schema/context/spring-context-3.0.xsd"
        default-lazy-init="true">   
    
    	<context:annotation-config />
    	
        <bean id="abstractXADataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close" abstract="true">
            <property name="xaDataSourceClassName" value="com.alibaba.druid.pool.xa.DruidXADataSource"/>
            <property name="xaProperties">
                <props>
                	<prop key="driverClassName">${jdbc.driverClassName}</prop>
                	
    				<prop key="initialSize">10</prop>
    				<prop key="minIdle">3</prop>
    				<prop key="maxActive">100</prop>
    				
    				<prop key="maxWait">60000</prop>
    				
    				<prop key="timeBetweenEvictionRunsMillis">60000</prop>
    				
    				<prop key="minEvictableIdleTimeMillis">300000</prop>
    				<prop key="validationQuery">SELECT 'x'</prop>
    				<prop key="testWhileIdle">true</prop>
    				<prop key="testOnBorrow">false</prop>
    				<prop key="testOnReturn">false</prop>
    				
    				<prop key="filters">stat</prop>
                </props>
            </property>
        </bean>
        
        <bean id="dataSourceOne" parent="abstractXADataSource">
            <property name="uniqueResourceName">
                <value>dataSourceOne</value>
            </property>
            <property name="xaProperties">
                <props>
                    <prop key="url">${jdbc.url}</prop>
                    <prop key="username">${jdbc.username}</prop>
                    <prop key="password">${jdbc.password}</prop>
                </props>
            </property>
        </bean>
        
        <bean id="dataSourceTwo" parent="abstractXADataSource">
            <property name="uniqueResourceName">
                <value>dataSourceTwo</value>
            </property>
            <property name="xaProperties">
                <props>
                    <prop key="url">${jdbc.two.url}</prop>
                    <prop key="username">${jdbc.two.username}</prop>
                    <prop key="password">${jdbc.two.password}</prop>
                </props>
            </property>
        </bean>
        
        
        <bean id="sqlSessionFactoryOne" class="org.mybatis.spring.SqlSessionFactoryBean">
            <property name="dataSource" ref="dataSourceOne"/>
            <property name="mapperLocations" value="classpath*:mapping/ds1/*.xml"/>
        </bean>
        
        <bean id="sqlSessionFactoryTwo" class="org.mybatis.spring.SqlSessionFactoryBean">
            <property name="dataSource" ref="dataSourceTwo"/>
            <property name="mapperLocations" value="classpath*:mapping/ds2/*.xml"/>
        </bean>
        
        
        <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
            <property name="basePackage" value="net.xiake6.dao.ds1"/>
            <property name="sqlSessionFactoryBeanName" value="sqlSessionFactoryOne"/>
        </bean>
    
        <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
            <property name="basePackage" value="net.xiake6.dao.ds2"/>
            <property name="sqlSessionFactoryBeanName" value="sqlSessionFactoryTwo"/>
        </bean>
    
        
        <bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" destroy-method="close">
            <property name="forceShutdown" value="false"/>
        </bean>
        <bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
            <property name="transactionTimeout" value="3000"/>
        </bean>
        
        <bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
            <property name="transactionManager">
                <ref bean="atomikosTransactionManager"/>
            </property>
            <property name="userTransaction">
                <ref bean="atomikosUserTransaction"/>
            </property>
            <property name="allowCustomIsolationLevels" value="true"/>
        </bean>
       
        
        <aop:config proxy-target-class="true">  
            <aop:advisor pointcut="execution(* *net.xiake6.service..*(..))" advice-ref="txAdvice" />  
        </aop:config>  
        <tx:advice id="txAdvice" transaction-manager="jtaTransactionManager">  
            <tx:attributes>  
                <tx:method name="insert*"  propagation="REQUIRED" read-only="false" rollback-for="*Exception"/>
                <tx:method name="add*" propagation="REQUIRED" read-only="false" rollback-for="*Exception" />  
                <tx:method name="save*"  propagation="REQUIRED"  read-only="false" rollback-for="*Exception" />    
                <tx:method name="delete*"  propagation="REQUIRED"  read-only="false" rollback-for="*Exception" />  
                <tx:method name="del*"  propagation="REQUIRED"  read-only="false" rollback-for="*Exception" />  
                <tx:method name="update*"  propagation="REQUIRED"  read-only="false" rollback-for="*Exception" />
                <tx:method name="select*"  propagation="REQUIRED"  read-only="true" />
                <tx:method name="query"  propagation="REQUIRED"  read-only="true" />
            </tx:attributes>  
        </tx:advice>
        
        
        <tx:annotation-driven transaction-manager="jtaTransactionManager" />
    </beans>

**jdbc.properties**

#mysql 6.*以上
    jdbc.driverClassName = com.mysql.cj.jdbc.Driver
    jdbc.url = jdbc:mysql://127.0.0.1:3306/test?serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8&pinGlobalTxToPhysicalConnection=true&useSSL=false
    jdbc.username =root
    jdbc.password =root
    
    jdbc.two.url = jdbc:mysql://127.0.0.1:3306/test?serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8&pinGlobalTxToPhysicalConnection=true&useSSL=false
    jdbc.two.username =root
    jdbc.two.password =root

**jta.properties**

com.atomikos.icatch.service=com.atomikos.icatch.standalone.UserTransactionServiceFactory
    com.atomikos.icatch.console_file_name=tm.release.out
    com.atomikos.icatch.log_base_name=tm.releaselog
    com.atomikos.icatch.tm_unique_name=com.atomikos.spring.jdbc.tm.release
    com.atomikos.icatch.console_log_level=INFO

**TestInsert.java**

@ContextConfiguration(value = {"classpath:spring-application-context.xml"})
    @RunWith(SpringJUnit4ClassRunner.class)
    public class TestInsert {
    	private Logger logger = LoggerFactory.getLogger(TestInsert.class);
    	@Autowired
    	private BatchInsertService batchInsertService;
    
    	@Test
    	public void insert(){
    		
    		long startTime = System.currentTimeMillis();
    		User user = new User();
    		user.setName("User_"+(int)(Math.random()*100));
    		user.setAge((int)(Math.random()*100));
    		
    		CustInfo info = new CustInfo();
    		info.setPhone("123456789"+(int)(Math.random()*100));
    		batchInsertService.insert(user,info);
    		
    		long endTime = System.currentTimeMillis();
    		logger.info("共耗时：{}毫秒",endTime -startTime);
    	}
    }

**BatchInsertService.java**

@Service
    public class BatchInsertService {
    	private Logger logger = LoggerFactory.getLogger(BatchInsertService.class);
    	@Autowired
    	private UserService userService;
    	@Autowired
    	private CustInfoService custInfoService;
    	@Transactional(rollbackFor= {Exception.class,RuntimeException.class})
    	public void insert(User user,CustInfo custInfo) {
    		int insertUser =  userService.insert(user);
    		logger.info("insertUser={}",insertUser);
    		int insertCustInfo = custInfoService.insert(custInfo);
    		logger.info("insertCustInfo={}",insertCustInfo);
    	}
    }

UserService.java

@Service
    public class UserService {
    	@Autowired
    	private UserMapper userMapper;
    	
    	public int insert(User record) {
    		int result = userMapper.insert(record);
    		return result;
    	}
    	
    }

CustInfoService.java

@Service
    public class CustInfoService {
    	@Autowired
    	private CustInfoMapper custInfoMapper;
    
    	public int insert(CustInfo record) {
    		int result = custInfoMapper.insert(record);
    		long now = System.currentTimeMillis();
    		// 模拟一个异常
    		if (now % 2 == 0) {
    			throw new RuntimeException("CustInfoMapper throws test insert exception");
    		}
    		return result;
    	}
    }

Mapper和Bean等就不列出来了，完成的示例工程在github：[https://github.com/fenglibin/DruidWithAtomikos][https_github.com_fenglibin_DruidWithAtomikos]

[https_github.com_fenglibin_DruidWithAtomikos]: https://github.com/fenglibin/DruidWithAtomikos