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

waynaqua2022年12月16日约 8 分钟

在作者之前的十二条后端开发经验分享，纯干货^[1] 文章中介绍的 优雅得Springboot + mybatis配置多数据源方式 里有很多小伙伴在评论区留言询问多个数据源同时在一个方法中使用时，事务是否会正常有效，这里作者 理论 + 实践 给大家解答一波，老规矩，附作者github地址：

• https://github.com/wayn111

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

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

下面给出代码验证：

主数据源配置

@Slf4j
@EnableTransactionManagement
@EnableAspectJAutoProxy
@Configuration
@MapperScan(basePackages ="ltd.newbee.mall.core.dao", sqlSessionFactoryRef ="masterSqlSessionFactory")
publicclassDb1DataSourceConfig{@Primary
@Bean
@ConfigurationProperties("spring.datasource.druid.master")
publicDataSourcemasterDataSource(DruidProperties druidProperties){
DruidDataSourcebuild=DruidDataSourceBuilder.create().build();
return druidProperties.dataSource(build);
}
/**
     * @param datasource 数据源
     * @return SqlSessionFactory
     * @Primary 默认SqlSessionFactory
     */
@Primary
@Bean(name ="masterSqlSessionFactory")
publicSqlSessionFactorymasterSqlSessionFactory(@Qualifier("masterDataSource")DataSource datasource,
Interceptor interceptor)throwsException{
MybatisSqlSessionFactoryBeanbean=newMybatisSqlSessionFactoryBean();
        bean.setDataSource(datasource);
// mybatis扫描xml所在位置
        bean.setMapperLocations(newPathMatchingResourcePatternResolver()
.getResources("classpath*:mapper/*.xml"));
        bean.setTypeAliasesPackage("ltd.**.core.entity");
        bean.setPlugins(interceptor);
GlobalConfigglobalConfig=newGlobalConfig();
GlobalConfig.DbConfigdbConfig=newGlobalConfig.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")
publicDataSourceTransactionManagermasterTransactionManager(@Qualifier("masterDataSource")DataSource dataSource){
returnnewDataSourceTransactionManager(dataSource);
}
}

从数据源配置

@Slf4j
@ConditionalOnProperty(value ="transactional.mode", havingValue ="seata")
@EnableTransactionManagement
@EnableAspectJAutoProxy
@Configuration
@MapperScan(basePackages ="ltd.newbee.mall.slave.dao", sqlSessionFactoryRef ="slaveSqlSessionFactory")
publicclassDb2DataSourceConfig{@Bean
@ConfigurationProperties("spring.datasource.druid.slave")
publicDataSourceslaveDataSource(DruidProperties druidProperties){
DruidDataSourcebuild=DruidDataSourceBuilder.create().build();
return druidProperties.dataSource(build);
}
/**
     * @param datasource 数据源
     * @return SqlSessionFactory
     * @Primary 默认SqlSessionFactory
     */
@Bean(name ="slaveSqlSessionFactory")
publicSqlSessionFactoryslaveSqlSessionFactory(@Qualifier("slaveDataSource")DataSource datasource,
Interceptor interceptor)throwsException{
MybatisSqlSessionFactoryBeanbean=newMybatisSqlSessionFactoryBean();
        bean.setDataSource(datasource);
// mybatis扫描xml所在位置
        bean.setMapperLocations(newPathMatchingResourcePatternResolver()
.getResources("classpath*:slavemapper/*.xml"));
        bean.setTypeAliasesPackage("ltd.**.slave.entity");
        bean.setPlugins(interceptor);
GlobalConfigglobalConfig=newGlobalConfig();
GlobalConfig.DbConfigdbConfig=newGlobalConfig.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")
publicDataSourceTransactionManagerslaveTransactionManager(@Qualifier("slaveDataSource")DataSource dataSource){
returnnewDataSourceTransactionManager(dataSource);
}
}

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

/**
 *  Springboot测试类
 */
@Slf4j
@SpringBootTest
@RunWith(SpringRunner.class)
publicclassMultiDataSourceTest{
@Autowired
privateMultiDataService multiDataService;
@Test
publicvoidtestRollback(){
        multiDataService.testRollback();
}
}
/**
 *  MultiDataService实现类
 */
@Slf4j
@Service
publicclassMultiDataServiceImplimplementsMultiDataService{
@Autowired
privateTbTable1Service tbTable1Service;
@Autowired
privateTbTable2Service tbTable2Service;
@Autowired
privatePlatformTransactionManager transactionManager;
@Override
publicvoidtestRollback(){
DefaultTransactionDefinitiontransactionDefinition=newDefaultTransactionDefinition();
TransactionStatustransaction= transactionManager.getTransaction(transactionDefinition);
try{
TbTable1tbTable1=newTbTable1();
            tbTable1.setName("test1");
// 插入table1表
booleansave1= tbTable1Service.save(tbTable1);
TbTable2tbTable2=newTbTable2();
            tbTable2.setName("test2");
// 插入table2表
booleansave2= tbTable2Service.save(tbTable2);
inti=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")
publicDataSourcedataSource(DruidProperties druidProperties){
DruidXADataSourcedataSource= druidProperties.dataSource(newDruidXADataSource());
        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");
AtomikosDataSourceBeanatomikosDataSourceBean=newAtomikosDataSourceBean();
        atomikosDataSourceBean.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
        atomikosDataSourceBean.setUniqueResourceName("master-xa");
        atomikosDataSourceBean.setXaDataSource(dataSource);
return atomikosDataSourceBean;
}

1. 添加 JtaTransactionManager

@Bean
publicJtaTransactionManager transactionManager()throwsException{
JtaTransactionManager transactionManager =newJtaTransactionManager();
UserTransactionManager userTransactionManager =newUserTransactionManager();
    userTransactionManager.setForceShutdown(true);
    userTransactionManager.setTransactionTimeout(3000);
    transactionManager.setUserTransaction(userTransactionManager);
    transactionManager.setAllowCustomIsolationLevels(true);
return transactionManager;
}

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

/**
 *  Springboot测试类
 */
@Slf4j
@SpringBootTest
@RunWith(SpringRunner.class)
publicclassMultiDataSourceTest{
@Autowired
privateMultiDataService multiDataService;
@Test
publicvoidjtaTestRollback(){
        multiDataService.jtaTestRollback();
}
}
/**
 *  MultiDataService实现类
 */
@Slf4j
@Service
publicclassMultiDataServiceImplimplementsMultiDataService{
@Autowired
privateTbTable1Service tbTable1Service;
@Autowired
privateTbTable2Service tbTable2Service;
@Autowired
privateJtaTransactionManager jtaTransactionManager;
@Override
publicvoidjtaTestRollback(){
DefaultTransactionDefinitiontransactionDefinition=newDefaultTransactionDefinition();
TransactionStatustransaction= jtaTransactionManager.getTransaction(transactionDefinition);
try{
TbTable1tbTable1=newTbTable1();
            tbTable1.setName("test1");
booleansave1= tbTable1Service.save(tbTable1);
TbTable2tbTable2=newTbTable2();
            tbTable2.setName("test2");
booleansave2= tbTable2Service.save(tbTable2);
inti=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，这里给出官网教程：https://seata.io/zh-cn/docs/ops/deploy-guide-beginner.html
io.seata seata-spring-boot-starter 1.5.2
2. 在 Springboot中引入seata最新依赖
3. 在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. 完成测试，代码如下：

/**
 *  Springboot测试类
 */
@Slf4j
@SpringBootTest
@RunWith(SpringRunner.class)
publicclassMultiDataSourceTest{
@Autowired
privateMultiDataService multiDataService;
@Test
publicvoidseataTestRollback(){
        multiDataService.seataTestRollback();
}
}
/**
 *  MultiDataService实现类
 */
@Slf4j
@Service
publicclassMultiDataServiceImplimplementsMultiDataService{
@Autowired
privateTbTable1Service tbTable1Service;
@Autowired
privateTbTable2Service tbTable2Service;
@GlobalTransactional
@Override
publicvoidseataTestRollback(){
        log.info("当前 XID: {}",RootContext.getXID());
TbTable1tbTable1=newTbTable1();
        tbTable1.setName("test1");
booleansave1= tbTable1Service.save(tbTable1);
TbTable2tbTable2=newTbTable2();
        tbTable2.setName("test2");
booleansave2= tbTable2Service.save(tbTable2);
inti=1/0;
}
}

如上代码，使用 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项目：https://github.com/wayn111/newbee-mall/tree/springboot2.7 分支下

欢迎大家点赞、关注、评论，想要跟作者沟通技术问题的话可以加我微信【waynaqua】，欢迎大家前来交流。

引用链接

[1] 十二条后端开发经验分享，纯干货: https://juejin.cn/post/7171770606940061727

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

下面给出代码验证：

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

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

Springboot 项目中可以启用 jta

引入 seata，添加XA协议支持

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

总结

引用链接

来自同一个声音

17 万 Star 的 Matt Pocock Skills 到底好不好用？我让两个 Subagent 真跑了一遍

GPT-5.6 Sol 出来后，我把 Superpowers 从“必装”名单里删了

你的 Superpowers 该更新了：6.0 给 Agent 上规矩

Codex 额度不够用：额度翻倍小技巧 + 白嫖长任务通道