多数据源事务处理-涉及分布式事务

左手的ㄟ右手 2023-10-03 22:13 8阅读 0赞

## 一. 数据源跨库但是不跨 `MySql` 实例 ##

这个形式就是数据源在同一个 `MySQL` 下，但是 `jdbc-url` 上的数据库配置不同，涉及多个数据库时，如果方法中发生异常，只有开启事务的数据源会发生回滚，其他数据源不会回滚。看到这里可能有点迷惑，什么是 `只有开启事务的数据源会发生回滚，其他数据源不会回滚？`

### 下面给出代码验证： ###

主数据源配置

@Slf4j
    @EnableTransactionManagement
    @EnableAspectJAutoProxy
    @Configuration
    @MapperScan(basePackages = "ltd.newbee.mall.core.dao", sqlSessionFactoryRef = "masterSqlSessionFactory")
    public class Db1DataSourceConfig {
    
        @Primary
        @Bean
        @ConfigurationProperties("spring.datasource.druid.master")
        public DataSource masterDataSource(DruidProperties druidProperties) {
            DruidDataSource build = DruidDataSourceBuilder.create().build();
            return druidProperties.dataSource(build);
        }
    
        /**
         * @param datasource 数据源
         * @return SqlSessionFactory
         * @Primary 默认SqlSessionFactory
         */
        @Primary
        @Bean(name = "masterSqlSessionFactory")
        public SqlSessionFactory masterSqlSessionFactory(@Qualifier("masterDataSource") DataSource datasource,
                                                         Interceptor interceptor) throws Exception {
            MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
            bean.setDataSource(datasource);
            // mybatis扫描xml所在位置
            bean.setMapperLocations(new PathMatchingResourcePatternResolver()
                    .getResources("classpath*:mapper/*.xml"));
            bean.setTypeAliasesPackage("ltd.**.core.entity");
            bean.setPlugins(interceptor);
            GlobalConfig globalConfig = new GlobalConfig();
            GlobalConfig.DbConfig dbConfig = new GlobalConfig.DbConfig();
            dbConfig.setLogicDeleteField("isDeleted");
            dbConfig.setLogicDeleteValue("1");
            dbConfig.setLogicNotDeleteValue("0");
            globalConfig.setDbConfig(dbConfig);
            bean.setGlobalConfig(globalConfig);
            log.info("masterDataSource 配置成功");
            return bean.getObject();
        }
    
        @Primary
        @Bean(name = "masterTransactionManager")
        public DataSourceTransactionManager masterTransactionManager(@Qualifier("masterDataSource") DataSource dataSource) {
            return new DataSourceTransactionManager(dataSource);
        }
    
    }
    复制代码

从数据源配置

@Slf4j
    @ConditionalOnProperty(value = "transactional.mode", havingValue = "seata")
    @EnableTransactionManagement
    @EnableAspectJAutoProxy
    @Configuration
    @MapperScan(basePackages = "ltd.newbee.mall.slave.dao", sqlSessionFactoryRef = "slaveSqlSessionFactory")
    public class Db2DataSourceConfig {
    
        @Bean
        @ConfigurationProperties("spring.datasource.druid.slave")
        public DataSource slaveDataSource(DruidProperties druidProperties) {
            DruidDataSource build = DruidDataSourceBuilder.create().build();
            return druidProperties.dataSource(build);
        }
    
    
        /**
         * @param datasource 数据源
         * @return SqlSessionFactory
         * @Primary 默认SqlSessionFactory
         */
        @Bean(name = "slaveSqlSessionFactory")
        public SqlSessionFactory slaveSqlSessionFactory(@Qualifier("slaveDataSource") DataSource datasource,
                                                        Interceptor interceptor) throws Exception {
            MybatisSqlSessionFactoryBean bean = new MybatisSqlSessionFactoryBean();
            bean.setDataSource(datasource);
            // mybatis扫描xml所在位置
            bean.setMapperLocations(new PathMatchingResourcePatternResolver()
                    .getResources("classpath*:slavemapper/*.xml"));
            bean.setTypeAliasesPackage("ltd.**.slave.entity");
            bean.setPlugins(interceptor);
            GlobalConfig globalConfig = new GlobalConfig();
            GlobalConfig.DbConfig dbConfig = new GlobalConfig.DbConfig();
            dbConfig.setLogicDeleteField("isDeleted");
            dbConfig.setLogicDeleteValue("1");
            dbConfig.setLogicNotDeleteValue("0");
            globalConfig.setDbConfig(dbConfig);
            bean.setGlobalConfig(globalConfig);
            log.info("slaveDataSource 配置成功");
            return bean.getObject();
        }
        
        @Bean(name = "slaveTransactionManager")
        public DataSourceTransactionManager slaveTransactionManager(@Qualifier("slaveDataSource") DataSource dataSource) {
            return new DataSourceTransactionManager(dataSource);
        }
    
    }
    复制代码

**划重点-上述代码在每个数据源中都配置了 `DataSourceTransactionManager`（事务管理器），并且在主配置中添加 `@Primary` 注解，表示默认事务管理器优先使用主数据源的事务管理器。** 下面给出测试代码：

/**
     *  Springboot测试类
     */
    @Slf4j
    @SpringBootTest
    @RunWith(SpringRunner.class)
    public class MultiDataSourceTest {
        @Autowired
        private MultiDataService multiDataService;
        @Test
        public void testRollback() {
            multiDataService.testRollback();
        }
    }
    /**
     *  MultiDataService实现类
     */
    @Slf4j
    @Service
    public class MultiDataServiceImpl implements MultiDataService {
        @Autowired
        private TbTable1Service tbTable1Service;
        @Autowired
        private TbTable2Service tbTable2Service;
        @Autowired
        private PlatformTransactionManager transactionManager;
        @Override
        public void testRollback() {
            DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
            TransactionStatus transaction = transactionManager.getTransaction(transactionDefinition);
            try {
                TbTable1 tbTable1 = new TbTable1();
                tbTable1.setName("test1");
                // 插入table1表
                boolean save1 = tbTable1Service.save(tbTable1);
                TbTable2 tbTable2 = new TbTable2();
                tbTable2.setName("test2");
                // 插入table2表
                boolean save2 = tbTable2Service.save(tbTable2);
                int i = 1 / 0;
                transactionManager.commit(transaction);
                Assert.isTrue(save1 && save2);
            } catch (Exception e) {
                log.info(e.getMessage(), e);
                transactionManager.rollback(transaction);
            }
        }
    }
    复制代码

执行结果：**table1表回滚成功，table2表回滚失败**。由此结果，对于 `只有开启事务的数据源会发生回滚，其他数据源不会回滚？` 我们的解释就是 `Spring` 中默认使用的事务管理器是使用主数据源配置还是从数据源配置由我们通过 `@Primary` 决定，当我们把 `@Primary` 切换在从数据源配置上，执行结果：**table2表回滚成功，table1表回滚失败**。那怎么解决这个问题？

### 当涉及到跨库或者跨 `MySQL` 实例，想要保证事务操作，我们这里先给出`XA`事务解决方案。附 `XA` 事务的说明： ###

> XA 是由 X/Open 组织提出的分布式事务规范，XA 规范主要定义了事务协调者（Transaction Manager）和资源管理器（Resource Manager）之间的接口。

事务协调者（Transaction Manager），因为 XA 事务是基于两阶段提交协议的，所以需要有一个协调者，来保证所有的事务参与者都完成了准备工作，也就是 2PC 的第一阶段。如果事务协调者收到所有参与者都准备好的消息，就会通知所有的事务都可以提交，也就是 2PC 的第二阶段。

资源管理器（Resource Manager），负责控制和管理实际资源，比如数据库。

**（划重点）XA 的 MySQL 实现使 MySQL 服务器能够充当资源管理器，在全局事务中处理 XA 事务。连接到 MySQL 服务器的客户端程序充当事务协调者**

**XA 事务的执行流程**

XA 事务是两阶段提交的一种实现方式，根据 2PC 的规范，XA 将一次事务分割成了两个阶段，即 Prepare 和 Commit 阶段。

**Prepare 阶段**，TM 向所有 RM 发送 prepare 指令，RM 接受到指令后，执行数据修改和日志记录等操作，然后返回可以提交或者不提交的消息给 TM。如果事务协调者 TM 收到所有参与者都准备好的消息，会通知所有的事务提交，然后进入第二阶段。

**Commit 阶段**，TM 接受到所有 RM 的 prepare 结果，如果有 RM 返回是不可提交或者超时，那么向所有 RM 发送 Rollback 命令；如果所有 RM 都返回可以提交，那么向所有 RM 发送 Commit 命令，完成一次事务操作。

### 下面给出两种基于 `XA` 事务的解决方案： ###

*  `Springboot` 项目中可以使用 `jta`，完成对 `XA` 协议的支持，缺点就是 `jta` 需要改造数据源配置
 *  `Springboot` 项目引入 `seata` ，`seata` 支持 `XA` 协议，且引入 `seata-spring-boot-starter` 依赖对业务无侵入，缺点需要引入 `seata-server` 降低了系统可用性

### `Springboot` 项目中可以启用 `jta` ###

1.  引入 `spring-boot-starter-jta-atomikos`

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jta-atomikos</artifactId>
    </dependency>
    复制代码

1.  修改**主从数据源** `DataSource` 配置，进行包装添加 `XA` 数据源支持，如下；

@Primary
        @Bean
        @ConfigurationProperties("spring.datasource.druid.master")
        public DataSource dataSource(DruidProperties druidProperties) {
            DruidXADataSource dataSource = druidProperties.dataSource(new DruidXADataSource());
            dataSource.setUrl("jdbc:mysql://localhost:3306/xxx?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8");
            dataSource.setUsername("root");
            dataSource.setPassword("");
            dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
            AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
            atomikosDataSourceBean.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
            atomikosDataSourceBean.setUniqueResourceName("master-xa");
            atomikosDataSourceBean.setXaDataSource(dataSource);
            return atomikosDataSourceBean;
        }
    复制代码

1.  添加 `JtaTransactionManager`

@Bean
    public JtaTransactionManager transactionManager() throws Exception {
        JtaTransactionManager transactionManager = new JtaTransactionManager();
        UserTransactionManager userTransactionManager = new UserTransactionManager();
        userTransactionManager.setForceShutdown(true);
        userTransactionManager.setTransactionTimeout(3000);
        transactionManager.setUserTransaction(userTransactionManager);
        transactionManager.setAllowCustomIsolationLevels(true);
        return transactionManager;
    }
    复制代码

1.  完成测试，代码如下：

/**
     *  Springboot测试类
     */
    @Slf4j
    @SpringBootTest
    @RunWith(SpringRunner.class)
    public class MultiDataSourceTest {
        @Autowired
        private MultiDataService multiDataService;
        @Test
        public void jtaTestRollback() {
            multiDataService.jtaTestRollback();
        }
    }
    /**
     *  MultiDataService实现类
     */
    @Slf4j
    @Service
    public class MultiDataServiceImpl implements MultiDataService {
        @Autowired
        private TbTable1Service tbTable1Service;
        @Autowired
        private TbTable2Service tbTable2Service;
        @Autowired
        private JtaTransactionManager jtaTransactionManager;
        @Override
        public void jtaTestRollback() {
            DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
            TransactionStatus transaction = jtaTransactionManager.getTransaction(transactionDefinition);
            try {
                TbTable1 tbTable1 = new TbTable1();
                tbTable1.setName("test1");
                boolean save1 = tbTable1Service.save(tbTable1);
                TbTable2 tbTable2 = new TbTable2();
                tbTable2.setName("test2");
                boolean save2 = tbTable2Service.save(tbTable2);
                int i = 1 / 0;
                jtaTransactionManager.commit(transaction);
                Assert.isTrue(save1 && save2);
            } catch (Exception e) {
                log.info(e.getMessage(), e);
                jtaTransactionManager.rollback(transaction);
            }
        }
    }
    复制代码

可以看到我们使用的是 `JtaTransactionManager`, 执行结果：**table1表回滚成功，table2表回滚成功**。验证OK

### 引入 `seata`，添加XA协议支持 ###

1.  下载安装启动 `seata-server`，这里给出官网教程：[seata.io/zh-cn/docs/…][seata.io_zh-cn_docs]
2.  在 Springboot中引入seata最新依赖

<dependency>
        <groupId>io.seata</groupId>
        <artifactId>seata-spring-boot-starter</artifactId>
        <version>1.5.2</version>
    </dependency>
    复制代码

1.  在yml文件中添加 `seata` 配置

seata:
      config:
        type: file
      registry:
        type: file
      application-id: newbeemall # Seata 应用编号，默认为 ${spring.application.name}
      tx-service-group: newbeemall-group # Seata 事务组编号，用于 TC 集群名
      # 服务配置项，对应 ServiceProperties 类
      service:
        # 虚拟组和分组的映射
        vgroup-mapping:
          newbeemall-group: default
        # 分组和 Seata 服务的映射
        grouplist:
          default: 127.0.0.1:8091
      data-source-proxy-mode: XA
      enabled: true
    复制代码

1.  完成测试，代码如下：

如上代码，使用 `seata` 时需要启用 `@GlobalTransactional` 注解，并且在事务中传递 `XID`（ `RootContext.getXID()`），执行结果：**table1表回滚成功，table2表回滚成功**。验证OK

## 二. 数据源分布在不同 `MySql` 实例 ##

当数据源分布在不同 `MySql` 实例时，这时候其实已经进入分布式事务的范畴，由上可知，`XA` 事务可以解决分布式环境下事务问题，也就是说上述最后两种解决方案都可以解决分布式事务问题，但是实际使用过程中，我们建议使用 `seata`，理由是他不仅支持 `XA` 事务还支持 `AT、Saga、TCC`事务模型。引入 `seata` 官网介绍

> Seata 是一款开源的分布式事务解决方案，致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式，为用户打造一站式的分布式解决方案。

## 总结 ##

关于多数据源事务的问题，不管跨不跨库其实都属于分布式事务的问题。推荐使用 `seata` 解决。

实践代码放在newbeemall项目：[github.com/wayn111/new…][github.com_wayn111_new] 分支下

![format_png][]

[seata.io_zh-cn_docs]: https://link.juejin.cn?target=https%3A%2F%2Fseata.io%2Fzh-cn%2Fdocs%2Fops%2Fdeploy-guide-beginner.html
[github.com_wayn111_new]: https://link.juejin.cn?target=https%3A%2F%2Fgithub.com%2Fwayn111%2Fnewbee-mall%2Ftree%2Fspringboot2.7
[format_png]: https://img-blog.csdnimg.cn/img_convert/b68db16aaaf6d5a8a6ce32efb1832da9.webp?x-oss-process=image/format,png