Mybatis-plus多数据源深度剖析

科技 06-17 来源：架构修炼者

Mybatis-plus多数据源深度剖析

应用场景

开发过程中，经常会遇到这两种场景，第一个是业务需求需要操作多个DB场景，比如：下单时，需要从用户库中查询用户信息，同时需要向订单库里插入一条订单；另一个是读写分离场景，这个大家熟悉的不能在熟了，就不用过多介绍了。这两个场景就是典型的多数据源访问。

多数据源实现设计的思考点

让我们自己实现多数据源访问，应该考虑哪些点呢？大家都好好想想，一定要动脑筋思考下！让我做的话，需要解决3个问题，分别是：

配置来源问题：多数据源的配置存储在什么位置？yml文件，配置中心，还是缓存？
创建和管理问题：多个数据源如何创建？如何管理？用什么存储？
使用和切换问题：这么多数据源如何使用？如何切换？ORM框架只允许使用一个数据源，如何和ORM框架集成？

思考完后，不一定非要自己开发，你遇到的问题，别人早就遇到了，并且都有现成的解决方案，可以拿来主义，没有在自己动手也不完，带着问题先去调研一波，走起！

业界多数据源实现方案

业界有2种实现方案，分别为：

AOP + ThreadLocal ，如：Mybatis-plus的多数据源(dynamic-datasource)；
语义解析，如：客户端侧：ShardingSphere-Jdbc，服务端侧：ShardingSphere-Proxy，阿里云、腾讯云proxy。

多数据源带来的问题

引入多数据源后，解决了多数据源访问的问题，同时也带来另外2个问题：

事务问题：对多数据源写操作时，如何保证数据的一致性，完整性？
多层嵌套切换问题(AOP方案)：如：serviceA--->ServiceB--->ServiceC，如何保证每层都使用自己的数据源？

Mybatis-plus多数据源深度剖析

业界多数据源方案有很多种，咱们这次主要对Mybatis-plus多数据源(dynamic-datasource)进行一次深度剖析，首先来看它有哪些特性，然后带着这些特性去看源码。

特性

支持数据源分组，2种负载均衡策略：轮询和随机
支持对JDBC连接的url，username，password加密 ENC()
支持无数据源启动，动态增加删除数据源
支持每个数据库独立初始化表结构schema和数据库database。
支持数据源延迟初始化
提供并简化对Druid，HikariCp，BeeCp，Dbcp2的快速集成
提供自定义数据源来源方案（如全从数据库加载）
支持多层数据源嵌套切换。
提供本地多数据源事务方案
提供使用 spel动态参数解析数据源方案。内置spel，session，header，支持自定义。

多数据源创建流程

查找入口类和多数据源对象

首先来看下多数据源创建的入口类：DynamicDataSourceAutoConfiguration，很多人都会问我，你咋找的？来我告诉你咋找的，从dynamic-datasource-spring-boot-starter项目命名上，就能看出它是个starter，只要是starter，它就有自动化配置类，直接看这个resources/META-INF/spring.factories文件。

找到入口类后，别犹豫点它，在里面你会看到这个方法：

public DataSource dataSource() {
    DynamicRoutingDataSource dataSource = new DynamicRoutingDataSource();
    dataSource.setPrimary(properties.getPrimary());
    dataSource.setStrict(properties.getStrict());
    dataSource.setStrategy(properties.getStrategy());
    dataSource.setP6spy(properties.getP6spy());
    dataSource.setSeata(properties.getSeata());
    return dataSource;
}

DynamicRoutingDataSource就是咱们的多数据源对象，虽然说是多数据源，实际上它就是一个数据源对象（DataSource），而它里面封装了多个数据源，对外暴露就是一个对象。这就解决了这个问题：ORM框架只允许使用一个数据源，如何和ORM框架集成？

分析多数据源对象(DynamicRoutingDataSource)

首先来看下它的属性：

// 分组数据源的分隔标识符
private static final String UNDERLINE = "_";
// 用来存放了配置的所有数据库
private final Map dataSourceMap = new ConcurrentHashMap<>();
// 用来存放分组数据库，例如：配置了slave_1,slave_2两个数据源，存在是slave--->GroupDataSource(slave_1,slave_2)
private final Map groupDataSources = new ConcurrentHashMap<>();
// 数据源提供者,用来加载并创建数据源
private List providers;
// 负载均衡策略，用来对分组数据源进行负载均衡
private Class<? extends DynamicDataSourceStrategy> strategy = LoadBalanceDynamicDataSourceStrategy.class;

看完属性，然后了解下数据源在什么时候创建的，入口：afterPropertiesSet方法。

//com.baomidou.dynamic.datasource.DynamicRoutingDataSource#afterPropertiesSet 
public void afterPropertiesSet() throws Exception {
        // 检查开启了配置但没有相关依赖
        checkEnv();
        // 添加并分组数据源
        Map dataSources = new HashMap<>(16);
        for (DynamicDataSourceProvider provider : providers) {
            // provider.loadDataSources()是创建数据源!!!!!!!!!!!!!!!!!!!!!
            dataSources.putAll(provider.loadDataSources());
        }
        for (Map.Entry dsItem : dataSources.entrySet()) {
            // 添加数据源到属性：dataSourceMap 和 groupDataSources中
            addDataSource(dsItem.getKey(), dsItem.getValue());
        }
      .............
    }

分析数据源提供者(YmlDynamicDataSourceProvider)

最关键的是provider.loadDataSources()：用来创建数据源的。来看下provider，它的默认实现类是YmlDynamicDataSourceProvider，从命名上就能看出来数据源是从yaml文件中加载数据源的。

public class YmlDynamicDataSourceProvider extends AbstractDataSourceProvider {

    /**
     * 所有数据源配置
     */
    private final Map dataSourcePropertiesMap;

    // 加载数据源并创建数据源
    @Override
    public Map loadDataSources() {
        return createDataSourceMap(dataSourcePropertiesMap);
    }
}

来看下它的抽象类AbstractDataSourceProvider，这个抽象类存在的意义是什么？思考下，还记得特性中这条吗？特性：提供自定义数据源来源方案（如全从数据库加载），这就是它存在的意义。来看下具体创建数据源的方法：createDataSourceMap。

// 默认数据源创建工厂，使用的工厂模式
@Autowired
private DefaultDataSourceCreator defaultDataSourceCreator;

protected Map createDataSourceMap(
    Map dataSourcePropertiesMap) {
    Map dataSourceMap = new HashMap<>(dataSourcePropertiesMap.size() * 2);
    for (Map.Entry item : dataSourcePropertiesMap.entrySet()) {
    String dsName = item.getKey();
    DataSourceProperty dataSourceProperty = item.getValue();
    String poolName = dataSourceProperty.getPoolName();
    if (poolName == null || "".equals(poolName)) {
    poolName = dsName;
}
    dataSourceProperty.setPoolName(poolName);
    //defaultDataSourceCreator.createDataSource:用来创建数据源的     
    dataSourceMap.put(dsName,       defaultDataSourceCreator.createDataSource(dataSourceProperty));
}
    return dataSourceMap;
}

分析数据源创建工厂(DefaultDataSourceCreator)

DefaultDataSourceCreator：默认数据源创建工厂，使用的工厂模式。来看下DefaultDataSourceCreator类，

public class DefaultDataSourceCreator {
    // 存放所有数据源创建器，例如：常见DruidDataSourceCreator，HikariDataSourceCreator
    private List creators;

    public DataSource createDataSource(DataSourceProperty dataSourceProperty) {
        DataSourceCreator dataSourceCreator = null;
        for (DataSourceCreator creator : this.creators) {
            // 判断该创建器是否支持，支持的话就有该创建器来创建。
            if (creator.support(dataSourceProperty)) {
                dataSourceCreator = creator;
                break;
            }
        }
        if (dataSourceCreator == null) {
            throw new IllegalStateException("creator must not be null,please check the DataSourceCreator");
        }
        return dataSourceCreator.createDataSource(dataSourceProperty);
    }

}

属性creators：存放所有数据源创建器，目前提供了6个数据源创建器，分别为：BasicDataSourceCreator，BeeCpDataSourceCreator，Dbcp2DataSourceCreator，DruidDataSourceCreator，HikariDataSourceCreator，JndiDataSourceCreator。

来看下顶层接口：DataSourceCreator，提供了2个方法：createDataSource和support

public interface DataSourceCreator {

    /**
     * 通过属性创建数据源
     *
     * @param dataSourceProperty 数据源属性
     * @return 被创建的数据源
     */
    DataSource createDataSource(DataSourceProperty dataSourceProperty);

    /**
     * 当前创建器是否支持根据此属性创建
     *
     * @param dataSourceProperty 数据源属性
     * @return 是否支持
     */
    boolean support(DataSourceProperty dataSourceProperty);
}

看来顶层接口，来看下抽象类：AbstractDataSourceCreator，这就用到了模板方法设计模式。

com.baomidou.dynamic.datasource.creator.AbstractDataSourceCreator#createDataSource

// 实现DataSourceCreator接口中的createDataSource方法
public DataSource createDataSource(DataSourceProperty dataSourceProperty) {
        // 加密用的公钥
        String publicKey = dataSourceProperty.getPublicKey();
        if (StringUtils.isEmpty(publicKey)) {
            publicKey = properties.getPublicKey();
            dataSourceProperty.setPublicKey(publicKey);
        }
        // 设置延迟加载
        Boolean lazy = dataSourceProperty.getLazy();
        if (lazy == null) {
            lazy = properties.getLazy();
            dataSourceProperty.setLazy(lazy);
        }
        // 对JDBC连接的url，username，password解密
        dataSourceInitEvent.beforeCreate(dataSourceProperty);
        // 需要子类实现的抽象方法
        DataSource dataSource = doCreateDataSource(dataSourceProperty);
        dataSourceInitEvent.afterCreate(dataSource);
        // 支持每个数据库独立初始化表结构schema和数据库database。
        this.runScrip(dataSource, dataSourceProperty);
        return wrapDataSource(dataSource, dataSourceProperty);
    }

模板方法中，处理了2个特性：支持对JDBC连接的url，username，password加密 ENC()和支持每个数据库独立初始化表结构schema和数据库database。

解密的相关类：EncDataSourceInitEvent，表结构初始化类：ScriptRunner，底层调用的spring框架的ResourcePatternResolver和DatabasePopulatorUtils，不在展开，感兴趣的话，自己去研究下，都很简单。

下面以常见Druid的创建器DruidDataSourceCreator，来分析一波，其他创建器自己去看源码，大同小异。来看下doCreateDataSource方法：

    public DataSource doCreateDataSource(DataSourceProperty dataSourceProperty) {
        DruidDataSource dataSource = new DruidDataSource();
        dataSource.setUsername(dataSourceProperty.getUsername());
        dataSource.setPassword(dataSourceProperty.getPassword());
        dataSource.setUrl(dataSourceProperty.getUrl());
        dataSource.setName(dataSourceProperty.getPoolName());
        String driverClassName = dataSourceProperty.getDriverClassName();
        if (!StringUtils.isEmpty(driverClassName)) {
            dataSource.setDriverClassName(driverClassName);
        }
        DruidConfig config = dataSourceProperty.getDruid();
        Properties properties = config.toProperties(gConfig);

        List proxyFilters = this.initFilters(dataSourceProperty, properties.getProperty("druid.filters"));
        dataSource.setProxyFilters(proxyFilters);

        dataSource.configFromPropety(properties);
        //连接参数单独设置
        dataSource.setConnectProperties(config.getConnectionProperties());
        // 全局配置和自身配置整合，就近原则。
        this.setParam(dataSource, config);
        // 是否延迟初始化
        if (Boolean.FALSE.equals(dataSourceProperty.getLazy())) {
            try {
                dataSource.init();
            } catch (SQLException e) {
                throw new ErrorCreateDataSourceException("druid create error", e);
            }
        }
        return dataSource;
    }

这个方法涉及了一个特性： 支持数据源延迟初始化，那什么时候初始化呢？是在获取getConnection时，初始化。

// com.alibaba.druid.pool.DruidDataSource#getConnection(long)
public DruidPooledConnection getConnection(long maxWaitMillis) throws SQLException {
        // 初始化
        init();

        if (filters.size() > 0) {
            FilterChainImpl filterChain = new FilterChainImpl(this);
            return filterChain.dataSource_connect(this, maxWaitMillis);
        } else {
            return getConnectionDirect(maxWaitMillis);
        }
    }

到此，多数据源创建就完事了，来总结一把。

多数据源路由Key查找和切换流程

分析AOP切面类(DynamicDataSourceAnnotationAdvisor)

从上面的业界多数据源实现方案一节，我们了解Mybatis-plus的是通过AOP+ThreadLocal实现的，那切面是啥呢？让我们再回到自动化配置类DynamicDataSourceAutoConfiguration，你肯定能找到这个bean配置。

@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
@Bean
@ConditionalOnProperty(prefix = DynamicDataSourceProperties.PREFIX + ".aop", name = "enabled", havingValue = "true", matchIfMissing = true)
public Advisor dynamicDatasourceAnnotationAdvisor(DsProcessor dsProcessor) {
    DynamicDatasourceAopProperties aopProperties = properties.getAop();
    DynamicDataSourceAnnotationInterceptor interceptor = new DynamicDataSourceAnnotationInterceptor(aopProperties.getAllowedPublicOnly(), dsProcessor);
    DynamicDataSourceAnnotationAdvisor advisor = new DynamicDataSourceAnnotationAdvisor(interceptor, DS.class);
    advisor.setOrder(aopProperties.getOrder());
    return advisor;
}

切面类就是DynamicDataSourceAnnotationAdvisor。切面就会有切点（pointcut）和通知（advice）,它是基于注解DS实现的AOP拦截，所以切点是@DS，通知是DynamicDataSourceAnnotationInterceptor，它的构造函数传入了一个DsProcessor对象，这个时用来处理路由key查找用的，后面会详细讲。

进入DynamicDataSourceAnnotationAdvisor类内部，我们能看到构建切点的方法（buildPointcut）：

private Pointcut buildPointcut() {
    // 匹配类上的@DS注解
    Pointcut cpc = new AnnotationMatchingPointcut(annotation, true);
    // 匹配方法上的@DS注解
    Pointcut mpc = new AnnotationMethodPoint(annotation);
    return new ComposablePointcut(cpc).union(mpc);
}

从上面可以看出，会同时查找类和方法有没有@DS注解。

分析AOP通知类(DynamicDataSourceAnnotationInterceptor)

让我们来看看通知类：DynamicDataSourceAnnotationInterceptor，它是实现了MethodInterceptor接口，首先来看看invoke方法。

public Object invoke(MethodInvocation invocation) throws Throwable {
    // 路由key查找
    String dsKey = determineDatasourceKey(invocation);
    // 方法执行之前，先把路由key放入到ThreadLocal>中, 切换数据源使用。
    DynamicDataSourceContextHolder.push(dsKey);
    try {
        return invocation.proceed();
    } finally {
        // 方法执行完后，从ThreadLocal>中移除路由key
        DynamicDataSourceContextHolder.poll();
    }
}

来看下路由key查找的方法：determineDatasourceKey(invocation)：

private String determineDatasourceKey(MethodInvocation invocation) {
        String key = dataSourceClassResolver.findKey(invocation.getMethod(), invocation.getThis());
    // 路由key以#开头的，将由dsProcessor来处理，否则，直接返回key。
        return key.startsWith(DYNAMIC_PREFIX) ? dsProcessor.determineDatasource(invocation, key) : key;
    }

还记得上面从构造函数中传进来的的dsProcessor对象嘛，在此处用到了，路由key以#开头的，将由dsProcessor来处理，否则，直接返回key。

以#开头的路由key处理器(DsProcessor)

路由key查找，官方目前提供了3种，分别为基于header的查找(DsHeaderProcessor)、基于session的查找(DsSessionProcessor)和基于表达式查找(DsSpelExpressionProcessor)。具体怎么实现的，自己看代码很简单。

然后来看下他们查找的顺序：

咱们可以扩展自己的查找处理器，只需实现DsProcessor接口就行，然后把自己的处理器添加进去就可以了。

具体如何添加？请参考这个方法com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceAutoConfiguration#dsProcessor

这就是上面所说的特性：提供使用 spel动态参数解析数据源方案。内置spel，session，header，支持自定义。

路由key的切换(DynamicDataSourceContextHolder)

来看下DynamicDataSourceContextHolder这个类，里面有个属性：

/**
     * 为什么要用链表存储(准确的是栈)
     *      * 为了支持嵌套切换，如ABC三个service都是不同的数据源
     * 其中A的某个业务要调B的方法，B的方法需要调用C的方法。一级一级调用切换，形成了链。
     * 传统的只设置当前线程的方式不能满足此业务需求，必须使用栈，后进先出。
     * 
     */
private static final ThreadLocal> LOOKUP_KEY_HOLDER = new NamedThreadLocal>("dynamic-datasource") {
        @Override
        protected Deque initialValue() {
            return new ArrayDeque<>();
        }
    };

路由key最终都会存在LOOKUP_KEY_HOLDER中，它是使用栈实现的。这就是上面所说的特性：支持多层数据源嵌套切换。

让我们来看看它怎么切换的，入口是获取连接getConnection方法。com.baomidou.dynamic.datasource.ds.AbstractRoutingDataSource#getConnection() ---》com.baomidou.dynamic.datasource.DynamicRoutingDataSource#determineDataSource

public DataSource determineDataSource() {
    // 从ThreadLocal中获取路由key，来切换数据源的。
    String dsKey = DynamicDataSourceContextHolder.peek();
    return getDataSource(dsKey);
}

到此，多数据源的路由key查找和切换就完成了，让我们来总结下：

自定义扩展点

我们来撸下有哪些位置我们可以扩展，下面是我总结这些扩展点，希望能帮助到你。

项目示例实战

yml文件添加多数据源配置

spring:
  application:
    name: demo
  datasource:
    dynamic:
      primary: master #设置默认的数据源或者数据源组,默认值即为master
      strict: false #严格匹配数据源,默认false. true未匹配到指定数据源时抛异常,false使用默认数据源
      datasource:
        master:
          url: jdbc:mysql://127.0.0.1:3306/test?useSSL=true&charset=utf8mb4&serverTimezone=Hongkong
          username: root
          password: 123456
          driver-class-name: com.mysql.cj.jdbc.Driver # 3.2.0开始支持SPI可省略此配置
        order:
          url: jdbc:mysql://127.0.0.1:3306/only_db_0?useSSL=true&charset=utf8mb4&serverTimezone=Hongkong
          username: root
          password: 123456
          driver-class-name: com.mysql.cj.jdbc.Driver

service类：

@Slf4j
@Service
public class TOrderServiceImpl extends ServiceImpl implements TOrderService {

    @Autowired
    private TOrderMapper tOrderMapper;

    @DS("master")
    @Override
    public Boolean insertMaster(TOrder tOrder) {
        int result = baseMapper.insert(tOrder);
        return result > 0 ? true : false;
    }

    @DS("order")
    @Override
    public Boolean insertOrder(TOrder tOrder) {
        int result = baseMapper.insert(tOrder);
        return result > 0 ? true : false;
    }
}

controller类：

@Slf4j
@RestController
@RequestMapping("order")
public class OrderController {

    @Autowired
    private TOrderService orderInfoService;

    @PostMapping("/insert")
    public String insert( @RequestBody TOrder request) throws InterruptedException {
        log.info("OrderInfoController.insert params:{}", JSON.toJSON(request));

        TenantContextHolder.setTenant(RandomUtil.randomNumbers(20));
        Boolean master = orderInfoService.insertMaster(request);
        Boolean order = orderInfoService.insertOrder(request);
        return "master=" + master + ",order=" + order;
    }
}

测试结果：