Atomikos详解：数据库XA规范与Atomikos使用与源码分析

一、认识2PC - 两阶段提交

1、理论

理论性的东西，懒得再打一遍了，贴在这了：
分布式事务详解【分布式事务的几种解决方案】彻底搞懂分布式事务
关键的两张图：
下图展示了2PC的两个阶段，分成功和失败两个情况阐明：
成功情况：

失败情况：

2、手撸XA-两阶段提交

（1）时序图

（2）代码实例

1. import com.mysql.cj.jdbc.JdbcConnection;
2. import com.mysql.cj.jdbc.MysqlXAConnection;
3. import com.mysql.cj.jdbc.MysqlXid;
4. import org.junit.jupiter.api.Test;
5. import org.springframework.boot.test.context.SpringBootTest;
7. import javax.transaction.xa.XAException;
8. import javax.transaction.xa.XAResource;
9. import javax.transaction.xa.Xid;
10. import java.sql.Connection;
11. import java.sql.DriverManager;
12. import java.sql.PreparedStatement;
13. import java.sql.SQLException;
15. @SpringBootTest
16. public class MysqlXaTest {
18.     @Test
19.     public void testXa() {
20.         try {
21.             //获取员工库的连接以及资源管理器
22.             JdbcConnection employeeConnection = (JdbcConnection) DriverManager.getConnection("jdbc:mysql://localhost:3306/employee", "root", "rootroot");
23.             MysqlXAConnection employeeXAConnection = new MysqlXAConnection(employeeConnection, true);
24.             XAResource employeeXaResource = employeeXAConnection.getXAResource();
26.             //获取的员工薪资库的连接以及资源管理器
27.             JdbcConnection salaryConnection = (JdbcConnection) DriverManager.getConnection("jdbc:mysql://localhost:3306/salary", "root", "rootroot");
28.             MysqlXAConnection salaryXAConnection = new MysqlXAConnection(salaryConnection, true);
29.             XAResource salaryXaResource = salaryXAConnection.getXAResource();
32.             // 全局事务id
33.             byte[] gtrid = "g00003".getBytes();
34.             // 分支事务id
35.             byte[] bqual = "b00001".getBytes();
36.             // 标识，一般是个固定值
37.             int formatId = 1;
39.             //开启员工插入的分支事务
40.             Xid employeeXid = new MysqlXid(gtrid, bqual, formatId);
41.             employeeXaResource.start(employeeXid, XAResource.TMNOFLAGS);
42.             PreparedStatement preparedStatement = employeeConnection.prepareStatement("insert into employee (name, sex, level) values ('小10', '女', '7')");
43.             preparedStatement.execute();
44.             employeeXaResource.end(employeeXid, XAResource.TMSUCCESS);
46.             //开启员工薪资的分支事务
47.             byte[] salaryBqual = "b00002".getBytes();
48.             Xid salaryXid = new MysqlXid(gtrid, salaryBqual, formatId);
49.             salaryXaResource.start(salaryXid, XAResource.TMNOFLAGS);
50.             PreparedStatement salaryPreparedStatement = salaryConnection.prepareStatement("insert into employee_salary (employee_id, salary) values ('12', 7000)");
51.             salaryPreparedStatement.execute();
52.             salaryXaResource.end(salaryXid, XAResource.TMSUCCESS);
54.             //第一阶段-准备阶段
55.             int employeePrepareResult = employeeXaResource.prepare(employeeXid);
56.             int salaryPrepareResult = salaryXaResource.prepare(salaryXid);
58.             //第二阶段-根据准备阶段的结果。判断是要执行commit还是rollback
59.             if (employeePrepareResult == XAResource.XA_OK && salaryPrepareResult == XAResource.XA_OK) {
60.                 employeeXaResource.commit(employeeXid, false);
61.                 salaryXaResource.commit(salaryXid, false);
62.             } else {
63.                 employeeXaResource.rollback(employeeXid);
64.                 salaryXaResource.rollback(salaryXid);
65.             }
66.         } catch (SQLException | XAException e) {
67.             throw new RuntimeException(e);
68.         }
69.     }
70. }

复制代码

3、认识JTA

JTA（Java Transaction API）：是Java平台上一个尺度API，用于管理和控制分布式事务的实行流程。
核心类：
javax.transaction.UserTransaction：袒露给应用使用，用来启动、提交、回滚事务。
javax.transaction.TransactionManager：提供给事务管理器的接口，用于协调和控制分布式事务的实行过程。
javax.transaction.XAResource：表示一个资源管理器，用于管理和操纵资源。
javax.transaction.Xid：用于唯一标识一个分布式事务。
4、今天的主角：Atomikos

Atomikos是一个开源的事务管理器，用于管理和控制分布式事务的实行过程。提供了一个可靠的、高性能的事务管明白决方案，可以与多种应用程序和数据库集成。
简单明白就是，Atomikos是可以集成在我们Java代码里面，和我们的业务代码绑定到同一个Java进程里面的一个事务管理器的框架，可以帮助我们业务程序去自行实现分布式事务。
Atomikos特点：支持分布式事务、支持多种web服务器、支持多种数据库、支持XA协议、提供高性能的事务管理。
Atomikos可以解决，在同一个应用下，毗连多个数据库，实现分布式事务。
5、2PC存在的问题

1、TM单点问题。TM挂掉之后，无法回滚和提交。
2、资源锁定的问题。资源锁定之后，TM挂掉无法回滚和提交。
3、性能瓶颈。资源锁定时间长。
4、数据不同等问题。commit时成功状态不同等就会造成数据不同等。

二、Atomikos使用

1、依赖+设置

1. <dependency>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-jta-atomikos</artifactId>
4. </dependency>

复制代码

1. server.port=8080
3. spring.employee-datasource.driverClassName = com.mysql.jdbc.Driver
4. spring.employee-datasource.jdbc-url = jdbc:mysql://localhost:3306/employee?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC
5. spring.employee-datasource.username = root
6. spring.employee-datasource.password = rootroot
8. spring.salary-datasource.driverClassName = com.mysql.jdbc.Driver
9. spring.salary-datasource.jdbc-url = jdbc:mysql://localhost:3306/salary?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC
10. spring.salary-datasource.username = root
11. spring.salary-datasource.password = rootroot
13. logging.level.com.atomikos = debug

复制代码

2、定义AtomikosDataSourceBean数据源

1. import com.atomikos.jdbc.AtomikosDataSourceBean;
2. import org.springframework.beans.factory.annotation.Value;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.context.annotation.Configuration;
6. import javax.sql.DataSource;
7. import java.util.Properties;
9. @Configuration
10. public class AtomikosDataSourceConfig {
12.     @Value("${spring.employee-datasource.jdbc-url}")
13.     private String employeeUrl;
15.     @Value("${spring.employee-datasource.username}")
16.     private String employeeUser;
18.     @Value("${spring.employee-datasource.password}")
19.     private String employeePassword;
21.     @Value("${spring.salary-datasource.jdbc-url}")
22.     private String salaryUrl;
24.     @Value("${spring.salary-datasource.username}")
25.     private String salaryUser;
27.     @Value("${spring.salary-datasource.password}")
28.     private String salaryPassword;
30.     /**
31.      * 定义两个数据源，分别对应两个数据库
32.      */
33.     @Bean(name = "employeeDataSource")
34.     public DataSource employeeDataSource(){
35.         AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
36.         atomikosDataSourceBean.setUniqueResourceName("employeeDataSource");
37.         atomikosDataSourceBean.setXaDataSourceClassName("com.mysql.cj.jdbc.MysqlXADataSource");
39.         Properties properties = new Properties();
40.         properties.setProperty("URL", employeeUrl);
41.         properties.setProperty("user", employeeUser);
42.         properties.setProperty("password", employeePassword);
43.         atomikosDataSourceBean.setXaProperties(properties);
44.         return atomikosDataSourceBean;
45.     }
47.     /**
48.      * 定义两个数据源，分别对应两个数据库
49.      */
50.     @Bean(name = "salaryDataSource")
51.     public DataSource salaryDataSource(){
52.         AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
53.         atomikosDataSourceBean.setUniqueResourceName("salaryDataSource");
54.         atomikosDataSourceBean.setXaDataSourceClassName("com.mysql.cj.jdbc.MysqlXADataSource");
56.         Properties properties = new Properties();
57.         properties.setProperty("URL", salaryUrl);
58.         properties.setProperty("user", salaryUser);
59.         properties.setProperty("password", salaryPassword);
60.         atomikosDataSourceBean.setXaProperties(properties);
61.         return atomikosDataSourceBean;
62.     }
63. }

复制代码

3、定义事务管理器JtaTransactionManager

1. import com.atomikos.icatch.jta.UserTransactionImp;
2. import com.atomikos.icatch.jta.UserTransactionManager;
3. import org.springframework.context.annotation.Bean;
4. import org.springframework.context.annotation.Configuration;
5. import org.springframework.context.annotation.DependsOn;
6. import org.springframework.transaction.PlatformTransactionManager;
7. import org.springframework.transaction.jta.JtaTransactionManager;
9. import javax.transaction.TransactionManager;
10. import javax.transaction.UserTransaction;
12. @Configuration
13. public class AtomikosConfig {
15.     // JTA的事务管理
16.     @Bean(name = "userTransaction")
17.     public UserTransaction userTransaction() {
18.         return new UserTransactionImp();
19.     }
21.     @Bean(name = "atomikosTransactionManager")
22.     public TransactionManager atomikosTransactionManager() {
23.         return new UserTransactionManager();
24.     }
26.     /**
27.      * 事务管理器
28.      */
29.     @Bean(name = "platformTransactionManager")
30.     @DependsOn({"userTransaction", "atomikosTransactionManager"})
31.     public PlatformTransactionManager transactionManager() {
32.         UserTransaction userTransaction = userTransaction();
33.         TransactionManager transactionManager = atomikosTransactionManager();
34.         return new JtaTransactionManager(userTransaction, transactionManager);
35.     }
36. }

复制代码

4、MyBatis设置

1. import lombok.SneakyThrows;
2. import org.apache.ibatis.session.SqlSessionFactory;
3. import org.mybatis.spring.SqlSessionFactoryBean;
4. import org.mybatis.spring.annotation.MapperScan;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. import javax.sql.DataSource;
11. @Configuration
12. @MapperScan(basePackages = "com.example.distributetransaction.dao", sqlSessionFactoryRef = "sqlSessionFactoryEmployee")
13. public class EmployeeMybatisConfig {
15.     @SneakyThrows
16.     @Bean
17.     public SqlSessionFactory sqlSessionFactoryEmployee(@Qualifier("employeeDataSource") DataSource dataSource) throws Exception {
18.         SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
19.         factoryBean.setDataSource(dataSource);
20.         return factoryBean.getObject();
21.     }
22. }

复制代码

1. import lombok.SneakyThrows;
2. import org.apache.ibatis.session.SqlSessionFactory;
3. import org.mybatis.spring.SqlSessionFactoryBean;
4. import org.mybatis.spring.annotation.MapperScan;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. import javax.sql.DataSource;
11. @Configuration
12. @MapperScan(basePackages = "com.example.distributetransaction.dao1", sqlSessionFactoryRef = "sqlSessionFactorySalary")
13. public class SalaryMybatisConfig {
15.     @SneakyThrows
16.     @Bean
17.     public SqlSessionFactory sqlSessionFactorySalary(@Qualifier("salaryDataSource") DataSource dataSource) throws Exception {
18.         SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
19.         factoryBean.setDataSource(dataSource);
20.         return factoryBean.getObject();
21.     }
22. }

复制代码

5、验证

1. @Transactional(rollbackFor = Exception.class)
2. public String join(EmployeeEntity employeeEntity) {
3.     //第一步，插入员工基础信息
4.     employeeDao.insertEmployee(employeeEntity);
5.     //第二步，插入员工薪资
6.     employeeSalaryDao.insertEmployeeSalary(employeeEntity.getId(), employeeEntity.getSalary());
8.     int i = 1 / 0;
9.     return "员工入职成功";
10. }

复制代码

三、Atomikos源码分析

1、@Transactional入口：TransactionInterceptor创建事务流程

• （1）Spring事务入口：@Transactional
  
• （2）TransactionInterceptor#invoke：Spring事务的署理拦截方法
  
• （3）TransactionAspectSupport#determineTransactionManager：确定事务管理器=>我们创建的JtaTransactionManager
  
• （4）TransactionAspectSupport#createTransactionIfNecessary：创建事务
  
• （5）AbstractPlatformTransactionManager#getTransaction：获取事务
  
• （6）JtaTransactionManager#doGetTransaction：获取事务，拿到JtaTransactionObject，里面封装了UserTransactionImp
  
• （7）JtaTransactionManager获取我们设置的UserTransactionImp
  
  
  
  
  
  
  

JtaTransactionManager#doGetTransaction：获取事务

2、启动事务

• （1）从AbstractPlatformTransactionManager#handleExistingTransaction调用AbstractPlatformTransactionManager#startTransaction开启事务
  
• （2）调用JtaTransactionManager#doBegin开启事务
  
• （3）调用JtaTransactionManager#doJtaBegin开启事务
  
• （4）调用UserTransactionImp#begin开启事务
  
• （5）最终调用的是UserTransactionManager#begin开启事务
  
• （6）调用TransactionManagerImp#begin()开启事务
  
• （7）调用CompositeTransactionManagerImp#createCompositeTransaction创建分布式事务
  
  
  
  
  
  
  
  
  
  
  
  
  

3、小总结：启动全局事务流程图

4、分支事务，业务流程实行过程

5、事务提交与回滚

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。