前言

可能由于业务上的某些需求，我们的系统中有时往往要连接多个数据库，这就产生了多数据源问题。

多数据源的情况下，一般我们要做到可以自动切换，此时会涉及到事务注解 Transactional 不生效问题和分布式事务问题。

关于多数据源方案，笔者在网上看过一些例子，然而大部分都是错误示例，根本跑不通，或者没办法兼容事务。

今天，我们就一点点来分析这些问题产生的根源和相应的解决方法。

多数据源

为了剧情的顺利开展，我们模拟的业务是创建订单和扣减库存。

所以，我们先创建订单表和库存表。注意，把他们分别放到两个数据库中。

CREATE TABLE `t_storage` (  `id` int(11) NOT NULL AUTO_INCREMENT,  `commodity_code` varchar(255) DEFAULT NULL,  `count` int(11) DEFAULT '0',  PRIMARY KEY (`id`),  UNIQUE KEY `commodity_code` (`commodity_code`)) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

CREATE TABLE `t_order` (  `id` bigint(16) NOT NULL,  `commodity_code` varchar(255) DEFAULT NULL,  `count` int(11) DEFAULT '0',  `amount` double(14,2) DEFAULT '0.00',  PRIMARY KEY (`id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;

1、数据库连接

通过YML文件先把两个数据库都配置一下。

spring:  datasource:    ds1:      jdbc_url: jdbc:mysql://127.0.0.1:3306/db1      username: root      password: root    ds2:      jdbc_url: jdbc:mysql://127.0.0.1:3306/db2      username: root      password: root

2、配置DataSource

我们知道，Mybatis执行一条SQL语句的时候，需要先获取一个Connection。这时候，就交由Spring管理器到DataSource中获取连接。

Spring中有个具有路由功能的DataSource，它可以通过查找键调用不同的数据源，这就是AbstractRoutingDataSource。

public abstract class AbstractRoutingDataSource{    //数据源的集合    @Nullable    private Map targetDataSources;//默认的数据源@Nullableprivate Object defaultTargetDataSource;//返回当前的路由键，根据该值返回不同的数据源@Nullableprotected abstract Object determineCurrentLookupKey();//确定一个数据源protected DataSource determineTargetDataSource() {//抽象方法 返回一个路由键        Object lookupKey = determineCurrentLookupKey();        DataSource dataSource = this.targetDataSources.get(lookupKey);return dataSource;    }}

可以看到，该抽象类的核心就是先设置多个数据源到Map集合中，然后根据Key可以获取不同的数据源。

那么，我们就可以重写这个determineCurrentLookupKey方法，它返回的是一个数据源的名称。

public class DynamicDataSource extends AbstractRoutingDataSource {    @Override    protected Object determineCurrentLookupKey() {        DataSourceType.DataBaseType dataBaseType = DataSourceType.getDataBaseType();        return dataBaseType;    }}

然后还需要一个工具类，来保存当前线程的数据源类型。

public class DataSourceType {

    public enum DataBaseType {        ds1, ds2    }    // 使用ThreadLocal保证线程安全    private static final ThreadLocal TYPE = new ThreadLocal();// 往当前线程里设置数据源类型public static void setDataBaseType(DataBaseType dataBaseType) {if (dataBaseType == null) {throw new NullPointerException();        }        TYPE.set(dataBaseType);    }// 获取数据源类型public static DataBaseType getDataBaseType() {        DataBaseType dataBaseType = TYPE.get() == null ? DataBaseType.ds1 : TYPE.get();return dataBaseType;    }}

这些都搞定之后，我们还需要把这个DataSource配置到Spring容器中去。下面这个配置类的作用如下：

• 创建多个数据源DataSource，ds1 和 ds2；

• 将ds1 和 ds2 数据源放入动态数据源DynamicDataSource；

• 将DynamicDataSource注入到SqlSessionFactory。

@Configurationpublic class DataSourceConfig {

    /**     * 创建多个数据源 ds1 和 ds2     * 此处的Primary，是设置一个Bean的优先级     * @return     */    @Primary    @Bean(name = "ds1")    @ConfigurationProperties(prefix = "spring.datasource.ds1")    public DataSource getDateSource1() {        return DataSourceBuilder.create().build();    }    @Bean(name = "ds2")    @ConfigurationProperties(prefix = "spring.datasource.ds2")    public DataSource getDateSource2() {        return DataSourceBuilder.create().build();    }

    /**     * 将多个数据源注入到DynamicDataSource     * @param dataSource1     * @param dataSource2     * @return     */    @Bean(name = "dynamicDataSource")    public DynamicDataSource DataSource(@Qualifier("ds1") DataSource dataSource1,                                        @Qualifier("ds2") DataSource dataSource2) {        Map targetDataSource = new HashMap<>();        targetDataSource.put(DataSourceType.DataBaseType.ds1, dataSource1);        targetDataSource.put(DataSourceType.DataBaseType.ds2, dataSource2);        DynamicDataSource dataSource = new DynamicDataSource();        dataSource.setTargetDataSources(targetDataSource);        dataSource.setDefaultTargetDataSource(dataSource1);return dataSource;    }/**     * 将动态数据源注入到SqlSessionFactory     * @param dynamicDataSource     * @return     * @throws Exception     */@Bean(name = "SqlSessionFactory")public SqlSessionFactory getSqlSessionFactory(@Qualifier("dynamicDataSource") DataSource dynamicDataSource)            throws Exception {        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();        bean.setDataSource(dynamicDataSource);        bean.setMapperLocations(                new PathMatchingResourcePatternResolver().getResources("classpath*:mapping/*.xml"));        bean.setTypeAliasesPackage("cn.youyouxunyin.multipledb2.entity");return bean.getObject();    }}

3、设置路由键

上面的配置都完成之后，我们还需要想办法动态的改变数据源的键值，这个就跟系统的业务相关了。

比如在这里，我们有两个Mapper接口，创建订单和扣减库存。

public interface OrderMapper {    void createOrder(Order order);}public interface StorageMapper {    void decreaseStorage(Order order);}

那么，我们就可以搞一个切面，在执行订单的操作时，切到数据源ds1，执行库存操作时，切到数据源ds2。

@Component@Aspectpublic class DataSourceAop {    @Before("execution(* cn.youyouxunyin.multipledb2.mapper.OrderMapper.*(..))")    public void setDataSource1() {        DataSourceType.setDataBaseType(DataSourceType.DataBaseType.ds1);    }    @Before("execution(* cn.youyouxunyin.multipledb2.mapper.StorageMapper.*(..))")    public void setDataSource2() {        DataSourceType.setDataBaseType(DataSourceType.DataBaseType.ds2);    }}

4、测试

现在就可以写一个Service方法，通过REST接口测试一下啦。

public class OrderServiceImpl implements OrderService {    @Override    public void createOrder(Order order) {        storageMapper.decreaseStorage(order);        logger.info("库存已扣减，商品代码:{}，购买数量:{}。创建订单中...",order.getCommodityCode(),order.getCount());        orderMapper.createOrder(order);    }}

不出意外的话，业务执行完成后，两个数据库的表都已经有了变化。

但此时，我们会想到，这两个操作是需要保证原子性的。所以，我们需要依赖事务，在Service方法上标注Transactional。

如果我们在createOrder方法上添加了Transactional注解，然后在运行代码，就会抛出异常。

### Cause: java.sql.SQLSyntaxErrorException: Table 'db2.t_order' doesn't exist; bad SQL grammar []; nested exception is java.sql.SQLSyntaxErrorException:     Table 'db2.t_order' doesn't exist] with root cause

这就说明，如果加上了 Spring 的事务，我们的数据源切换不过去了。这又是咋回事呢？

事务模式，为啥不能切换数据源

要想搞清楚原因，我们就得来分析分析如果加上了Spring事务，它又干了哪些事情呢 ？

我们知道，Spring的自动事务是基于AOP实现的。在调用包含事务的方法时，会进入一个拦截器。

public class TransactionInterceptor{    public Object invoke(MethodInvocation invocation) throws Throwable {        //获取目标类        Class> targetClass = AopUtils.getTargetClass(invocation.getThis());        //事务调用        return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);    }}

1、创建事务

在这里面呢，首先就是开始创建一个事务。

protected Object doGetTransaction() {    //DataSource的事务对象    DataSourceTransactionObject txObject = new DataSourceTransactionObject();    //设置事务自动保存    txObject.setSavepointAllowed(isNestedTransactionAllowed());    //给事务对象设置ConnectionHolder    ConnectionHolder conHolder = TransactionSynchronizationManager.getResource(obtainDataSource());    txObject.setConnectionHolder(conHolder, false);    return txObject;}

在这一步，重点是给事务对象设置了ConnectionHolder属性，不过此时还是为空。

2、开启事务

接下来，就是开启一个事务，这里主要是通过ThreadLocal将资源和当前的事务对象绑定，然后设置一些事务状态。

protected void doBegin(Object txObject, TransactionDefinition definition) {

    Connection con = null;    //从数据源中获取一个连接    Connection newCon = obtainDataSource().getConnection();    //重新设置事务对象中的connectionHolder，此时已经引用了一个连接    txObject.setConnectionHolder(new ConnectionHolder(newCon), true);    //将这个connectionHolder标记为与事务同步    txObject.getConnectionHolder().setSynchronizedWithTransaction(true);    con = txObject.getConnectionHolder().getConnection();    con.setAutoCommit(false);    //激活事务活动状态    txObject.getConnectionHolder().setTransactionActive(true);    //将connection holder绑定到当前线程，通过threadlocal    if (txObject.isNewConnectionHolder()) {        TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());    }    //事务管理器，激活事务同步状态    TransactionSynchronizationManager.initSynchronization();}

3、执行Mapper接口

开启事务之后，就开始执行目标类真实方法。在这里，就会开始进入Mybatis的代理对象。。哈哈，框架嘛，就各种代理。

我们知道，Mybatis在执行SQL的之前，需要先获取到SqlSession对象。

public static SqlSession getSqlSession(SqlSessionFactory sessionFactory, ExecutorType executorType,                PersistenceExceptionTranslator exceptionTranslator) {

    //从ThreadLocal中获取SqlSessionHolder，第一次获取不到为空    SqlSessionHolder holder = TransactionSynchronizationManager.getResource(sessionFactory);

    //如果SqlSessionHolder为空，那也肯定获取不到SqlSession;    //如果SqlSessionHolder不为空，直接通过它来拿到SqlSession    SqlSession session = sessionHolder(executorType, holder);    if (session != null) {        return session;    }    //创建一个新的SqlSession    session = sessionFactory.openSession(executorType);    //如果当前线程的事务处于激活状态，就将SqlSessionHolder绑定到ThreadLocal    registerSessionHolder(sessionFactory, executorType, exceptionTranslator, session);    return session;}

拿到SqlSession之后，就开始调用Mybatis的执行器，准备执行SQL语句。在执行SQL之前呢，当然需要先拿到Connection连接。

public Connection getConnection() throws SQLException {    //通过数据源获取连接    //比如我们配置了多数据源，此时还会正常切换    if (this.connection == null) {        openConnection();    }    return this.connection;}

我们看openConnection方法，它的作用就是从数据源中获取一个Connection连接。如果我们配置了多数据源，此时是可以正常切换的。如果加了事务，之所以没有切换数据源，是因为第二次调用时，this.connection != null，返回的还是上一次的连接。

这是因为，在第二次获取SqlSession的时候，当前线程是从ThreadLocal中拿到的，所以不会重复获取Connection连接。

至此，在多数据源情况下，如果加了Spring事务，不能动态切换数据源的原因，我们应该都明白了。

在这里，笔者插播一道面试题：

• Spring是如何保证事务的？

那就是将多个业务操作，放到同一个数据库连接中，一起提交或回滚。

• 怎么做到，都在一个连接中呢？

这里就是各种ThreadlLocal的运用，想办法将数据库资源和当前事务绑定到一起。

事务模式，怎么支持切换数据源

上面我们已经把原因搞清楚了，接下来就看怎么支持它动态切换数据源。

其他配置都不变的情况下，我们需要创建两个不同的sqlSessionFactory。

@Bean(name = "sqlSessionFactory1")public SqlSessionFactory sqlSessionFactory1(@Qualifier("ds1") DataSource dataSource){    return createSqlSessionFactory(dataSource);}

@Bean(name = "sqlSessionFactory2")public SqlSessionFactory sqlSessionFactory2(@Qualifier("ds2") DataSource dataSource){    return createSqlSessionFactory(dataSource);}

然后自定义一个CustomSqlSessionTemplate，来代替Mybatis中原有的sqlSessionTemplate，把上面定义的两个SqlSessionFactory注入进去。

@Bean(name = "sqlSessionTemplate")public CustomSqlSessionTemplate sqlSessionTemplate(){    Map sqlSessionFactoryMap = new HashMap<>();    sqlSessionFactoryMap.put("ds1",factory1);    sqlSessionFactoryMap.put("ds2",factory2);    CustomSqlSessionTemplate customSqlSessionTemplate = new CustomSqlSessionTemplate(factory1);    customSqlSessionTemplate.setTargetSqlSessionFactorys(sqlSessionFactoryMap);    customSqlSessionTemplate.setDefaultTargetSqlSessionFactory(factory1);return customSqlSessionTemplate;}

在定义的CustomSqlSessionTemplate中，其他都一样，主要看获取SqlSessionFactory的方法。

public class CustomSqlSessionTemplate extends SqlSessionTemplate {    @Override    public SqlSessionFactory getSqlSessionFactory() {        //当前数据源的名称        String currentDsName = DataSourceType.getDataBaseType().name();        SqlSessionFactory targetSqlSessionFactory = targetSqlSessionFactorys.get(currentDsName);        if (targetSqlSessionFactory != null) {            return targetSqlSessionFactory;        } else if (defaultTargetSqlSessionFactory != null) {            return defaultTargetSqlSessionFactory;        }        return this.sqlSessionFactory;    }}

在这里，重点就是我们可以根据不同的数据源获取不同的SqlSessionFactory。如果SqlSessionFactory不一样，那么在获取SqlSession的时候，就不会在ThreadLocal中拿到，从而每次都是新的SqlSession对象。

既然SqlSession也不一样，那么在获取Connection连接的时候，每次都会去动态数据源中去获取。

原理就是这么个原理，我们来走一把。

修改完配置之后，我们把Service方法加上事务的注解，此时数据也是可以正常更新的。

@Transactional@Overridepublic void createOrder(Order order) {    storageMapper.decreaseStorage(order);    orderMapper.createOrder(order);}

可以切换数据源只是第一步，我们需要的保证可以保证事务操作。假如在上面的代码中，库存扣减完成，但是创建订单失败，库存是不会回滚的。因为它们分别属于不同的数据源，根本不是同一个连接。

XA协议分布式事务

要解决上面那个问题，我们只能考虑XA协议。

关于XA协议是啥，笔者不再过多的描述。我们只需知道，MySQL InnoDB存储引擎是支持XA事务的。

那么XA协议的实现，在Java中叫做Java Transaction Manager，简称JTA。

如何实现JTA呢？我们借助Atomikos框架，先引入它的依赖。

<dependency>    <groupId>org.springframework.bootgroupId>    <artifactId>spring-boot-starter-jta-atomikosartifactId>    <version>2.2.7.RELEASEversion>dependency>

然后，只需把DataSource对象改成AtomikosDataSourceBean。

public DataSource getDataSource(Environment env, String prefix, String dataSourceName){    Properties prop = build(env,prefix);    AtomikosDataSourceBean ds = new AtomikosDataSourceBean();    ds.setXaDataSourceClassName(MysqlXADataSource.class.getName());    ds.setUniqueResourceName(dataSourceName);    ds.setXaProperties(prop);    return ds;}

这样配完之后，获取Connection连接的时候，拿到的其实是MysqlXAConnection对象。在提交或者回滚的时候，走的就是MySQL的XA协议了。

public void commit(Xid xid, boolean onePhase) throws XAException {    //封装 XA COMMIT 请求    StringBuilder commandBuf = new StringBuilder(300);    commandBuf.append("XA COMMIT ");    appendXid(commandBuf, xid);    try {        //交给MySQL执行XA事务操作        dispatchCommand(commandBuf.toString());    } finally {        this.underlyingConnection.setInGlobalTx(false);    }}

通过引入Atomikos和修改DataSource，在多数据源情况下，即便业务操作中间发生错误，多个数据库也是可以正常回滚的。

另外一个问题，是否应该使用XA协议？

XA协议看起来看起来比较简单，但它也有一些缺点。比如：

• 性能问题，所有参与者在事务提交阶段处于同步阻塞状态，占用系统资源，容易导致性能瓶颈，无法满足高并发场景；

• 如果协调者存在单点故障问题，如果协调者出现故障，参与者将一直处于锁定状态；

• 主从复制可能产生事务状态不一致。

在MySQL官方文档中也列举了一些XA协议的限制项：

https://dev.mysql.com/doc/refman/8.0/en/xa-restrictions.html

另外，笔者在实际的项目里，其实也没有用过，通过这样的方式来解决分布式事务问题，此例仅做可行性方案探讨。

总结

本文通过引入SpringBoot+Mybatis的多数据源场景，分析了如下问题：

• 多数据源的配置和实现；

• Spring事务模式，多数据源不生效的原因和解决方法；

• 多数据源，基于XA协议的分布式事务实现。

由于篇幅有限，本文示例不包含所有的代码。如有需要，请到GitHub自取。

https://github.com/taoxun/multipledb2.git

后台回复 学习资料 领取学习视频

如有收获，点个在看，诚挚感谢