Spring事务管理（二）分布式事务管理之JTA与链式事务

什么是分布式事务

跨库的事务就属于分布式事务，比如对两个库的不同表同时修改和同时rollback等。

上一节中，我们只是演示了单个库(数据源)的事务处理。这一节主要讲如何处理多个数据源的事务。

为什么多数据源下不能使用普通事务来处理呢?

我想很多人都有这个问题，打个比方，分库分表后有个数据库A和数据库B，A中有抢票记录，B中有票数记录。当我们完成抢票功能，需要在B减少票数的同时在A中增加记录。但是如果有下面的代码发生：

  1. @Transactional

  2. public void multiDBTX(){

  3.    B.reduce(ticketId);

  4.    if (true){

  5.        throw new RuntimeException("throw new exception");

  6.    }

  7.    A.save(result);

  8. }

我在B扣除票数后抛出异常，然后执行A库添加记录。

如果没有分布式事务处理，则结果就是B票数扣除，但A没有保存记录。也就是出错后B并没有进行事务回滚。

那问题来了，怎么才能实现我们的要求呢。

分布式事务原则

CAP定理

web无法同时满足以下三点：

  1. 一致性: 所有数据变动都是同步的

  2. 可用性: 每个操作都必须有预期的响应

  3. 分区容错性: 出现单个节点无法可用，系统依然正常对外提供服务

BASE理论

BASE理论是对CAP中的一致性和可用性进行一个权衡的结果。核心思想是即使无法做到强一致性，但可以使用一些技术手段达到最终一致。

  1. Basically Available（基本可用）:允许系统发生故障时，损失一部分可用性。

  2. Soft state（软状态）：允许数据同步存在延迟。

  3. Eventually consistent（最终一致性）： 不需要保持强一致性，最终一致即可。

那如何来实现分布式事务管理呢？

分布式事务管理实践

1. JTA实现

事务有效的屏蔽了底层事务资源，使应用可以以透明的方式参入到事务处理中，但是与本地事务相比，XA 协议的系统开销大

在这里我先带大家走出一个误解，你在网上搜JTA一般都是分布式事务用它，但是它就是用来做分布式事务的吗？不是的，我在上文说过，JTA只是Java实现XA事务的一个规范，我们在第一节 Spring事务管理（一）快速入门中用到的事务，都可以叫JTA事务管理。下面主要说JTA实现分布式事务管理：

这里我们会用到Atomikos事务管理器，它是一个开源的事务管理器，实现了XA的一种分布式事务处理并可以嵌入到你的SpringBoot当中。

拓展：什么是XA

基本上所有的数据库都会支持XA事务，百度百科上说法：XA协议由Tuxedo首先提出的，并交给X/Open组织，作为资源管理器（数据库）与事务管理器的接口标准。简单的说，它是事务的标准，JTA也是它标准的java实现。

1.1 导入pom

 

  1. <dependency>

  2.    <groupId>org.springframework.boot</groupId>

  3.    <artifactId>spring-boot-starter-jta-atomikos</artifactId>

  4. </dependency>

1.2 设置数据源

SpringBoot设置多数据源这里只说下思路(重点还是说事务实现)：

  1. 在 application.yml中配置多数据源配置。

  2. 写配置类加载配置并放入DataSource并设置事务：

 

 

  1. @Configuration

  2. @DependsOn("transactionManager")

  3. @EnableJpaRepositories(basePackages = "com.fantj.repository.user", entityManagerFactoryRef = "userEntityManager", transactionManagerRef = "transactionManager")

  4. @EnableConfigurationProperties(UserDatasourceProperties.class)

  5. public class UserConfig {

  6.  

  7.    @Autowired

  8.    private JpaVendorAdapter jpaVendorAdapter;

  9.  

  10.    // 这里注入 dataSource信息的类

  11.    @Autowired

  12.    private UserDatasourceProperties userDatasourceProperties;

  13.  

  14.    @Bean(name = "userDataSource")

  15.    public DataSource userDataSource() {

  16.        // 给XADataSource 设置 DataSource 属性

  17.        MysqlXADataSource mysqlXaDataSource = new MysqlXADataSource();

  18.        mysqlXaDataSource.setURL(userDatasourceProperties.getUrl());

  19.        mysqlXaDataSource.setUser(userDatasourceProperties.getUser());

  20.        mysqlXaDataSource.setPassword(userDatasourceProperties.getPassword());

  21.        mysqlXaDataSource.setPinGlobalTxToPhysicalConnection(true);

  22.        // 创建 Atomiko， 并将 mysql的XA交给JTA管理

  23.        AtomikosDataSourceBean xaDataSource = new AtomikosDataSourceBean();

  24.        xaDataSource.setXaDataSource(mysqlXaDataSource);

  25.        // 设置唯一资源名

  26.        xaDataSource.setUniqueResourceName("datasource2");

  27.        return xaDataSource;

  28.    }

  29.  

  30.    @Bean(name = "userEntityManager")

  31.    @DependsOn("transactionManager")

  32.    public LocalContainerEntityManagerFactoryBean userEntityManager() throws Throwable {

  33.  

  34.        HashMap<String, Object> properties = new HashMap<String, Object>();

  35.        properties.put("hibernate.transaction.jta.platform", AtomikosJtaPlatform.class.getName());

  36.        properties.put("javax.persistence.transactionType", "JTA");

  37.  

  38.        LocalContainerEntityManagerFactoryBean entityManager = new LocalContainerEntityManagerFactoryBean();

  39.        // 给工厂bean设置 资源加载属性

  40.        entityManager.setJtaDataSource(userDataSource());

  41.        entityManager.setJpaVendorAdapter(jpaVendorAdapter);

  42.        entityManager.setPackagesToScan("com.fantj.pojo.user");

  43.        entityManager.setPersistenceUnitName("userPersistenceUnit");

  44.        entityManager.setJpaPropertyMap(properties);

  45.        return entityManager;

  46.    }

  47.  

  48. }

这只是一个数据源的配置，第二个数据源的配置也类似，注意不能同Entity同Repository，映射放在不同包下实现。 两个都返回LocalContainerEntityManagerFactoryBean它便会交给@Transaction去管理，两个数据源配置完后。这样的代码B将会回滚。

 

  1. @Transactional

  2. public void multiDBTX(){

  3.    B.reduce(ticketId);

  4.    if (true){

  5.        throw new RuntimeException("throw new exception");

  6.    }

  7.    A.save(result);

  8. }

1.3 JTA缺点

因为JTA采用两阶段提交方式，第一次是预备阶段，第二次才是正式提交。当第一次提交出现错误，则整个事务出现回滚，一个事务的时间可能会较长，因为它要跨越多个 数据库 多个数据资源的的操作，所以在性能上可能会造成吞吐量低。而且，它只能用在单个服务内。一个完善的JTA事务还需要同时考虑很多元素，这只是个示例。

2. 链式事务管理

链式事务就是声明一个ChainedTransactionManager 将所有的数据源事务按顺序放到该对象中，则事务会按相反的顺序来执行事务。

网上发现了一个链式事务管理的处理顺序，总结的很到位。

 

  1. 1.start message transaction

  2. 2.receive message

  3. 3.start database transaction

  4. 4.update database

  5. 5.commit database transaction

  6. 6.commit message transaction   ##当这一步出现错误时，上面的因为已经commit，所以不会rollback

可以看到，从345可以看到，它后拿到的事务先提交，这就导致如果1出错，则不会进行数据回滚。跟Spring的同步事务差不多，同步事务也是这种特性。

下面我会测试这个性质。

为了方便，我拿JdbcTemplate来测试该事务。

DBConfig.java

配置DataSource以及返回Template新实例和链式事务配置。

 

  1. /**

  2. * DB配置类

  3. */

  4. @Configuration

  5. public class DBConfig {

  6.    /**

  7.     * user-DB配置

  8.     */

  9.    @Bean

  10.    @Primary

  11.    @ConfigurationProperties(prefix = "spring.datasource.user")

  12.    public DataSourceProperties userDataSourceProperties(){

  13.        return new DataSourceProperties();

  14.    }

  15.  

  16.    @Bean

  17.    @Primary

  18.    public DataSource userDataSource(){

  19.        return userDataSourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();

  20.    }

  21.  

  22.    @Bean

  23.    public JdbcTemplate userJdbcTemplate(@Qualifier("userDataSource") DataSource userDataSource){

  24.        return new JdbcTemplate(userDataSource);

  25.    }

  26.    /**

  27.     * result-DB配置

  28.     */

  29.    @Bean

  30.    @ConfigurationProperties(prefix = "spring.datasource.result")

  31.    public DataSourceProperties resultDataSourceProperties(){

  32.        return new DataSourceProperties();

  33.    }

  34.  

  35.    @Bean

  36.    public DataSource resultDataSource(){

  37.        return resultDataSourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();

  38.    }

  39.  

  40.    @Bean

  41.    public JdbcTemplate resultJdbcTemplate(@Qualifier("resultDataSource") DataSource resultDataSource){

  42.        return new JdbcTemplate(resultDataSource);

  43.    }

  44.    /**

  45.     * 链式事务配置

  46.     */

  47.    @Bean

  48.    public PlatformTransactionManager transactionManager(){

  49.        DataSourceTransactionManager userTM = new DataSourceTransactionManager(userDataSource());

  50.        DataSourceTransactionManager resultTM = new DataSourceTransactionManager(resultDataSource());

  51.        return new ChainedTransactionManager(userTM,resultTM);

  52.    }

  53. }

transactionManager()方法实现了链式事务配置，注意我放置的顺序先userTM后resultTM，所以事务应该是先拿到 userTM然后拿到 resultTM然后提交 resultTM最后提交 userTM，也就是说，如果我在提交user事务的时候出错，此时result相关的事务已经提交完成，所以result数据是不能回滚的。

2.1 测试

 

  1. @RequestMapping("")

  2. @Transactional

  3. public void testTX(){

  4. //        resultJdbcTemplate.execute("insert into result values(68,6,6)");

  5.    userJdbcTemplate.execute("insert into user values (6,'FantJ',23,'男')");

  6.    if (true){

  7.        throw new RuntimeException("yes , throw one exception");

  8.    }

  9.    resultJdbcTemplate.execute("insert into result values(66,6,6)");

  10. //        userJdbcTemplate.execute("insert into user values (8,'FantJ',23,'男')");

  11. }

两个数据库没有内容。

控制台精简后的日志：

 

  1. Creating new transaction with name [springbootjtamultidb.jtamulti.JtaMultiApplicationTests.testTX]:

  2. Creating new transaction with name [springbootjtamultidb.jtamulti.JtaMultiApplicationTests.testTX]:

  3. Began transaction (1) for test context [[email protected] testClass = ...

  4. Executing SQL statement [insert into user values (6,'FantJ',23,'男')]

  5. Initiating transaction rollback

  6. Rolling back JDBC transaction on Connection [[email protected] wrapping [email protected]]

  7. Releasing JDBC Connection [[email protected] wrapping com.mysql.cj.jdbc.

  8. Initiating transaction rollback

  9. Rolling back JDBC transaction on Connection [[email protected] wrapping [email protected]]

  10. Releasing JDBC Connection [[email protected] wrapping [email protected]] after transaction

  11. Resuming suspended transaction after completion of inner transaction

  12. Rolled back transaction for test: [[email protected] testClass = JtaMultiApplicationTests, testInstance = [email protected],

Spring事务管理（二）分布式事务管理之JTA与链式事务

Spring事务管理（二）分布式事务管理之JTA与链式事务Spring事务管理（二）分布式事务管理之JTA与链式事务

其实只要是两个dao操作中间出错或者第一个dao操作之前出错，事务都能正常回滚。如果result操作再前，user操作再后，user操作完抛出异常，也能回滚事务，原因上文有讲。

 

  1. @RequestMapping("")

  2. @Transactional

  3. public void testTX(){

  4.    resultJdbcTemplate.execute("insert into result values(68,6,6)");

  5. //        userJdbcTemplate.execute("insert into user values (6,'FantJ',23,'男')");

  6. //        if (true){

  7. //            throw new RuntimeException("yes , throw one exception");

  8. //        }

  9. //        resultJdbcTemplate.execute("insert into result values(66,6,6)");

  10.    userJdbcTemplate.execute("insert into user values (8,'FantJ',23,'男')");

  11.    if (true){

  12.        throw new RuntimeException("yes , throw one exception");

  13.    }

  14. }

这段代码也能正常回滚，结果我就不贴了。(浪费大家精力)

2.2 验证第二个事务不能回滚的情况

重要的事情再重复一遍:注意我放置的顺序先userTM后resultTM，所以事务应该是先拿到 userTM然后拿到 resultTM然后提交 resultTM最后提交 userTM，也就是说，如果我在提交user事务的时候出错，此时result相关的事务已经提交完成，所以result数据是不能回滚的。

 

  1. 代码和之前的一样，需要在事务提交的方法中打断点

  2. @RequestMapping("")

  3. @Transactional

  4. public void testTX(){

  5.    resultJdbcTemplate.execute("insert into result values(68,6,6)");

  6.    userJdbcTemplate.execute("insert into user values (8,'FantJ',23,'男')");

  7. }

Spring事务管理（二）分布式事务管理之JTA与链式事务

注意拦截到断点时，先放行一个commit，也就是result事务的commit，然后拦截到第二个commit请求时，关闭user所在的数据库，然后放行。

下面是将第一个commit请求放行后的控制台日志：

 

  1. Creating new transaction with name [springbootjtamultidb.jtamulti.controller.MainController.

  2. Acquired Connection [[email protected] wrapping [email protected]] for JDBC transaction

  3. Switching JDBC Connection [[email protected] wrapping [email protected]] to manual commit

  4. Executing SQL statement [insert into result values(68,6,6)]

  5. Executing SQL statement [insert into user values (8,'FantJ',23,'男')]

  6. Initiating transaction commit

  7. Committing JDBC transaction on Connection [[email protected] wrapping [email protected]]

  8. Releasing JDBC Connection [[email protected] wrapping [email protected]] after transaction

  9. Resuming suspended transaction after completion of inner transaction

注意倒数第三行日志，它出现证明我们第一个result的sql事务已经提交，此时你刷新数据库数据已经更新了，但是我们断点并还没有放行，user的事务还没有提交，我把user的数据库源关闭，再放行，可以看到，result已经有数据，user没有数据，此时result并没有进行回滚，这是链式事务的缺点。

Spring事务管理（二）分布式事务管理之JTA与链式事务

谈谈使用环境

JTA优缺点

JTA它的缺点是二次提交，事务时间长，数据锁的时间太长，性能比较低。

优点是强一致性，对于数据一致性要求很强的业务很有利，而且可以用于微服务。

链式/同步事务优缺点

优点： 比JTA轻量，能满足大部分事务需求，也是强一致性。

缺点： 只能单机玩，不能用于微服务，事务依次提交后提交的事务若出错不能回滚。

它两的比较

  1. JTA重，Chained轻。

  2. JTA能用于微服务分布式事务，Chained只能用于单机分布式事务。

事实上我们处理分布式事务都要求做到最终一致性。就是你刚开始我不需要保持你的数据一致，你中间可以出错，但是我能保证最终数据是一致的。

相关文章：

猜你喜欢
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode