Spring Boot+Atomikos进行多数据源的分布式事务管理详解和实例

0.前言

背景： 一直零散的使用着Spring Boot 的各种组件和特性，从未系统性的学习和总结，本次借着这个机会搞一波。共同学习，一起进步。哈哈

Atomikos是一个易用、可靠、开放源码的事务管理器，它可以用于管理分布式事务，尤其在微服务架构中非常实用。它支持JTA（Java Transaction API）规范，能够与各种JTA兼容的资源管理器（如数据库和消息队列）配合使用。

1. 分布式事务：当一个业务操作需要修改多个资源（如多个数据库或消息队列）时，我们需要保证这些修改操作的原子性，即它们要么全部成功，要么全部失败。这就是分布式事务。

2. JTA：Java Transaction API（JTA）是Java平台的一个事务规范，定义了用户和事务管理器以及事务管理器和资源管理器之间的接口。Atomikos作为一个事务管理器，就是遵循JTA规范的。

3. XA协议：XA协议是分布式事务的一个重要协议，它定义了全局事务ID、分支事务ID等概念，以及如何协调分支事务的接口。Atomikos支持XA协议。

Atomikos还提供了自动恢复、故障转移等高级特性，以进一步提高分布式事务的可靠性。

1.参考文档

Spring Boot 提供了一个用于整合 Atomikos 的 starter，名为 spring-boot-starter-jta-atomikos。它是 Spring Boot 提供的一系列 “starter” 依赖之一

1. Spring Boot 官方文档
   Spring Boot 官方文档的 “Spring Boot Features” 部分有一个 “Working with JTA” 的小节，其中提到了如何使用 spring-boot-starter-jta-atomikos。链接：https://docs.spring.io/spring-boot/docs/2.5.3/reference/htmlsingle/#boot-features-jta
   

2. Atomikos 官方文档
   虽然 Atomikos 的官方文档并没有专门介绍 spring-boot-starter-jta-atomikos，但它提供了一些关于如何使用 Atomikos 的教程，你可以参考这些教程来理解 spring-boot-starter-jta-atomikos 是如何工作的。链接：https://www.atomikos.com/Documentation/SpringBootIntegration

2.基础介绍

Atomikos 是一个提供分布式事务管理的开源事务管理器。将它们结合使用，可以在 Spring Boot 应用程序中实现分布式事务的管理。

在 Spring Boot 中使用 Atomikos，通常需要进行以下步骤：

1. 引入 Atomikos 依赖：
   首先，在 Maven 或 Gradle 构建文件中添加 Atomikos 的依赖项。可以添加 atomikos-transactions-spring-boot-starter 依赖，它是 Atomikos 与 Spring Boot 集成的起点。

2. 配置数据源：
   在 Spring Boot 应用程序中，你需要配置多个数据源。可以使用 Spring Boot 的自动配置功能，根据配置文件或属性来配置数据源。你可以使用任何支持 Atomikos 的数据源，如 Atomikos 提供的 AtomikosDataSourceBean 或其他第三方数据源。

3. 配置 Atomikos 事务管理器：
   在 Spring Boot 应用程序中，你需要配置 Atomikos 事务管理器。可以通过在配置类中创建 JtaTransactionManager 实例，并将其与 Atomikos 的 UserTransactionManager 和 TransactionManager 关联起来。这样，Spring 将使用 Atomikos 事务管理器来管理分布式事务。

4. 配置 JTA 事务管理器：
   为了使 Spring Boot 应用程序能够使用 Atomikos 进行分布式事务管理，你需要配置 JTA 事务管理器。可以使用 Spring Boot 的自动配置功能，根据配置文件或属性来配置 JTA 事务管理器。

5. 在方法上添加 @Transactional 注解：
   在需要进行事务管理的方法上，添加 Spring 的 @Transactional 注解。这将告诉 Spring 在方法执行期间启动和提交事务，并回滚事务（如果发生异常）。

使用 Spring Boot+Atomikos 的原理如下：

1. Spring Boot 提供了自动配置功能，可以根据配置文件或属性来自动配置数据源和事务管理器。

2. Atomikos 是一个独立的事务管理器，它提供了 JTA（Java Transaction API）的实现，可以处理分布式事务。

3. 在 Spring Boot 中，你配置了多个数据源，并使用 Atomikos 的数据源实现（如 AtomikosDataSourceBean）来实现分布式事务。

4. 当你在方法上添加了 @Transactional 注解时，Spring Boot 会使用 Atomikos 的事务管理器来管理事务。

5. 当方法执行时，事务管理器会协调各个数据源的事务，并在方法执行完成后根据事务的状态来提交或回滚事务。

总结来说，Spring Boot+Atomikos 的原理是利用 Spring Boot 的自动配置功能来配置多个数据源和事务管理器，并使用 Atomikos 的事务管理器实现分布式事务的管理。这样，你可以在 Spring Boot 应用程序中使用 @Transactional 注解来管理分布式事务。

3.步骤

确保你的数据库支持XA事务，否则无法使用Atomikos。

1. 添加依赖到你的pom.xml文件:

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>
<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId>
</dependency>

2. 配置数据源及其对应的JPA实体管理器和事务管理器:

在配置类中，我们需要分别为每个数据源创建DataSource、LocalContainerEntityManagerFactoryBean和JpaTransactionManager。这里假设有两个数据源db1和db2：

@Configuration
@EnableTransactionManagement
public class AtomikosConfig {@Bean@Primary@ConfigurationProperties(prefix = "spring.datasource.db1")public DataSource dataSource1() {return new AtomikosDataSourceBean();}@Bean@ConfigurationProperties(prefix = "spring.datasource.db2")public DataSource dataSource2() {return new AtomikosDataSourceBean();}@Bean@Primarypublic LocalContainerEntityManagerFactoryBean entityManagerFactory1() {HibernateJpaVendorAdapter jpaVendorAdapter = new HibernateJpaVendorAdapter();jpaVendorAdapter.setGenerateDdl(true);LocalContainerEntityManagerFactoryBean factory = new LocalContainerEntityManagerFactoryBean();factory.setJpaVendorAdapter(jpaVendorAdapter);factory.setPackagesToScan("com.example.package1");factory.setDataSource(dataSource1());factory.setPersistenceUnitName("persistenceUnit1");return factory;}@Beanpublic LocalContainerEntityManagerFactoryBean entityManagerFactory2() {HibernateJpaVendorAdapter jpaVendorAdapter = new HibernateJpaVendorAdapter();jpaVendorAdapter.setGenerateDdl(true);LocalContainerEntityManagerFactoryBean factory = new LocalContainerEntityManagerFactoryBean();factory.setJpaVendorAdapter(jpaVendorAdapter);factory.setPackagesToScan("com.example.package2");factory.setDataSource(dataSource2());factory.setPersistenceUnitName("persistenceUnit2");return factory;}
}

注意，这里使用了@Primary注解来标记主数据源和对应的实体管理器。

3. Spring Boot+MyBatis集成Atomikos

如果你的项目里同时使用了Spring Boot，MyBatis和Atomikos 。 两个数据源定义两个SqlSessionFactory和两个事务管理器，每个SqlSessionFactory和DataSourceTransactionManager都关联了一个特定的数据源。请注意，在我们定义SqlSessionFactory时，指定了mapper文件的路径，这是必需的，以便MyBatis知道如何将SQL语句映射到你的Java对象。你需要根据你的项目结构来修改这些路径。

@Configuration
@EnableTransactionManagement
public class AtomikosConfig {// 配置第一个数据源@Bean@Primary@ConfigurationProperties(prefix = "spring.datasource.db1")public DataSource dataSource1() {return new AtomikosDataSourceBean();}// 配置第二个数据源@Bean@ConfigurationProperties(prefix = "spring.datasource.db2")public DataSource dataSource2() {return new AtomikosDataSourceBean();}// 配置第一个SqlSessionFactory@Bean@Primarypublic SqlSessionFactory sqlSessionFactory1() throws Exception {SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();sqlSessionFactoryBean.setDataSource(dataSource1());sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:mapper/db1/*.xml"));return sqlSessionFactoryBean.getObject();}// 配置第二个SqlSessionFactory@Beanpublic SqlSessionFactory sqlSessionFactory2() throws Exception {SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();sqlSessionFactoryBean.setDataSource(dataSource2());sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:mapper/db2/*.xml"));return sqlSessionFactoryBean.getObject();}// 配置第一个事务管理器@Beanpublic DataSourceTransactionManager transactionManager1() {return new DataSourceTransactionManager(dataSource1());}// 配置第二个事务管理器@Beanpublic DataSourceTransactionManager transactionManager2() {return new DataSourceTransactionManager(dataSource2());}
}

4. 在application.properties文件中配置数据源和JPA属性：

spring.datasource.db1.unique-resource-name=datasource1
spring.datasource.db1.xa-data-source-class-name=com.mysql.cj.jdbc.MysqlXADataSource
spring.datasource.db1.xa-properties.databaseName=db1
spring.datasource.db1.xa-properties.url=jdbc:mysql://localhost:3306/db1
spring.datasource.db1.xa-properties.user=root
spring.datasource.db1.xa-properties.password=password
spring.datasource.db1.pool-size=5spring.datasource.db2.unique-resource-name=datasource2
spring.datasource.db2.xa-data-source-class-name=com.mysql.cj.jdbc.MysqlXADataSource
spring.datasource.db2.xa-properties.databaseName=db2
spring.datasource.db2.xa-properties.url=jdbc:mysql://localhost:3306/db2
spring.datasource.db2.xa-properties.user=root
spring.datasource.db2.xa-properties.password=password
spring.datasource.db2.pool-size=5

1. 创建Atomikos的UserTransaction和TransactionManager实例。

2. 创建要参与分布式事务的资源（如数据库连接）的XADataSource实例，并将它们注册到Atomikos。

3. 调用UserTransaction的begin方法开始事务。

4. 通过XADataSource获取资源连接，进行业务操作。

5. 调用UserTransaction的commit方法提交事务，或调用rollback方法回滚事务。

4.使用示例

接下来，我们将创建一个简单的服务来演示如何使用在两个数据库上进行分布式事务。这个服务将在两个数据库上进行数据的添加操作。

首先创建两个实体类，分别对应两个数据库的表：

@Entity
@Table(name = "test1")
public class Test1 {@Id@GeneratedValue(strategy = GenerationType.IDENTITY)private Long id;//...
}@Entity
@Table(name = "test2")
public class Test2 {@Id@GeneratedValue(strategy = GenerationType.IDENTITY)private Long id;//...
}

然后创建对应的Repository：

public interface Test1Repository extends JpaRepository<Test1, Long> {}public interface Test2Repository extends JpaRepository<Test2, Long> {}

接下来创建处理这两个数据库操作的Service：

@Service
public class TestService {private final Test1Repository test1Repository;private final Test2Repository test2Repository;public TestService(Test1Repository test1Repository, Test2Repository test2Repository) {this.test1Repository = test1Repository;this.test2Repository = test2Repository;}@Transactionalpublic void addData() {Test1 test1 = new Test1();//...test1Repository.save(test1);Test2 test2 = new Test2();//...test2Repository.save(test2);}
}

在addData方法中，我们在Test1和Test2两个表中分别添加数据。由于addData方法添加了@Transactional注解，所以这两个添加数据的操作会在同一个事务中执行。如果在添加数据到Test2表时出现了错误，那么添加数据到Test1表的操作也会被回滚。

在Controller或者其他层调用addData方法，例如：

@RestController
public class TestController {private final TestService testService;public TestController(TestService testService) {this.testService = testService;}@PostMapping("/addData")public ResponseEntity<String> addData() {testService.addData();return ResponseEntity.ok().body("Data added successfully");}
}

5.底层原理

Atomikos 的底层原理是基于 JTA 规范，通过事务管理器协调和管理分布式事务，使用两阶段提交协议保证事务的一致性，提供日志和恢复机制用于持久化和恢复事务状态，以及使用资源适配器与各个资源进行交互。这些组件和机制共同提供了可靠的分布式事务管理功能。

1. JTA 实现：
   Atomikos 实现了 JTA 规范，它提供了 javax.transaction 包中定义的接口和类的实现。这些接口包括 UserTransaction、TransactionManager 和 Transaction 等。Atomikos 利用这些接口和类来进行事务的管理和控制。

2. 事务管理器（Transaction Manager）：
   Atomikos 提供了一个事务管理器，用于协调和管理分布式事务。事务管理器负责事务的创建、启动、提交、回滚和状态管理等。它是 Atomikos 的核心组件，负责处理多个资源参与的分布式事务。

3. 本地事务管理：
   Atomikos 还支持本地事务管理，即仅在单个数据库或资源上执行的事务。对于本地事务，Atomikos 利用底层资源的本地事务管理功能，例如 JDBC 的本地事务或 JMS 的本地事务。

4. 分布式事务管理：
   对于涉及多个资源的分布式事务，Atomikos 使用两阶段提交（Two-Phase Commit，2PC）协议来保证事务的一致性。在这个协议中，事务管理器与各个资源管理器进行通信，确保所有资源都准备好提交事务，并在所有资源都准备好后，进行事务的提交操作。

5. 日志和恢复：
   Atomikos 还提供了日志和恢复机制，用于处理事务的持久化和恢复。它通过将事务的状态和操作记录到日志中，以确保事务的持久性。在系统故障或崩溃后，Atomikos 可以使用日志进行事务的恢复，并保证事务的一致性。

6. 资源适配器（Resource Adapter）：
   Atomikos 使用资源适配器来与各个资源进行交互，例如数据库、消息队列等。资源适配器负责管理资源的连接、事务的参与和操作的执行等。Atomikos 提供了一些内置的资源适配器，同时也支持自定义资源适配器。