springboot starter多数据源使用dynamic datasource springboot starter实现多数据源及源码分析

时间:2021-09-10 0x2015 人气:7

想了解使用dynamic datasource springboot starter实现多数据源及源码分析的相关内容吗，0x2015在本文为您仔细讲解springboot starter多数据源的相关知识和一些Code实例，欢迎阅读和指正，我们先划重点：dynamic,datasource,springboot,starter多数据源，下面大家一起来学习吧。

简介

前两篇博客介绍了用基本的方式做多数据源，可以应对一般的情况，但是遇到一些复杂的情况就需要扩展下功能了，比如：动态增减数据源、数据源分组，纯粹多库读写分离一主多从、从其他数据库或者配置中心读取数据源等等。其实就算没有这些需求，使用这个实现多数据源也比之前使用AbstractRoutingDataSource要便捷的多

dynamic-datasource-spring-boot-starter 是一个基于springboot的快速集成多数据源的启动器。

github: https://github.com/baomidou/dynamic-datasource-spring-boot-starter

文档： https://github.com/baomidou/dynamic-datasource-spring-boot-starter/wiki

它跟mybatis-plus是一个生态圈里的，很容易集成mybatis-plus

特性：

数据源分组，适用于多种场景纯粹多库读写分离一主多从混合模式。
内置敏感参数加密和启动初始化表结构schema数据库database。
提供对Druid，Mybatis-Plus，P6sy，Jndi的快速集成。
简化Druid和HikariCp配置，提供全局参数配置。
提供自定义数据源来源接口(默认使用yml或properties配置)。
提供项目启动后增减数据源方案。
提供Mybatis环境下的纯读写分离方案。
使用spel动态参数解析数据源，如从session，header或参数中获取数据源。（多租户架构神器）
提供多层数据源嵌套切换。（ServiceA >>> ServiceB >>> ServiceC，每个Service都是不同的数据源）
提供不使用注解而使用正则或 spel 来切换数据源方案（实验性功能）。
基于seata的分布式事务支持。

实操

先把坐标丢出来

<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
    <version>3.1.0</version>
</dependency>

下面抽几个用的比较多的应用场景介绍

基本使用

使用方法很简洁，分两步走
一：通过yml配置好数据源
二：service层里面在想要切换数据源的方法上加上@DS注解就行了，也可以加在整个service层上，方法上的注解优先于类上注解

spring:
  datasource:
    dynamic:
      primary: master #设置默认的数据源或者数据源组,默认值即为master
      strict: false #设置严格模式,默认false不启动. 启动后在未匹配到指定数据源时候回抛出异常,不启动会使用默认数据源.
      datasource:
        master:
          url: jdbc:mysql://127.0.0.1:3306/dynamic
          username: root
          password: 123456
          driver-class-name: com.mysql.jdbc.Driver
        db1:
          url: jdbc:gbase://127.0.0.1:5258/dynamic
          username: root
          password: 123456
          driver-class-name: com.gbase.jdbc.Driver

这就是两个不同数据源的配置，接下来写service代码就行了

# 多主多从
spring:
  datasource:
    dynamic:
      datasource:
        master_1:
        master_2:
        slave_1: 
        slave_2: 
        slave_3:

如果是多主多从，那么就用数据组名称_xxx，下划线前面的就是数据组名称，相同组名称的数据源会放在一个组下。切换数据源时，可以指定具体数据源名称，也可以指定组名然后会自动采用负载均衡算法切换

# 纯粹多库（记得设置primary）
spring:
  datasource:
    dynamic:
      datasource:
        db1:
        db2:
        db3: 
        db4: 
        db5:

纯粹多库，就一个一个往上加就行了

@Service
@DS("master")
public class UserServiceImpl implements UserService {

  @Autowired
  private JdbcTemplate jdbcTemplate;

  public List<Map<String, Object>> selectAll() {
    return jdbcTemplate.queryForList("select * from user");
  }
  
  @Override
  @DS("db1")
  public List<Map<String, Object>> selectByCondition() {
    return jdbcTemplate.queryForList("select * from user where age >10");
  }
}

注解	结果
没有@DS	默认数据源
@DS(“dsName”)	dsName可以为组名也可以为具体某个库的名称

在这里插入图片描述

通过日志可以发现我们配置的多数据源已经被初始化了，如果切换数据源也会看到打印日子的
是不是很便捷，这是官方的例子

集成druid连接池

<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.1.22</version>
</dependency>

首先引入依赖

spring:
  autoconfigure:
    exclude: com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceAutoConfigure

再排除掉druid原生的自动配置

spring:
  datasource: #数据库链接相关配置
    dynamic:
      druid: #以下是全局默认值，可以全局更改
        #监控统计拦截的filters
        filters: stat
        #配置初始化大小/最小/最大
        initial-size: 1
        min-idle: 1
        max-active: 20
        #获取连接等待超时时间
        max-wait: 60000
        #间隔多久进行一次检测，检测需要关闭的空闲连接
        time-between-eviction-runs-millis: 60000
        #一个连接在池中最小生存的时间
        min-evictable-idle-time-millis: 300000
        validation-query: SELECT 'x'
        test-while-idle: true
        test-on-borrow: false
        test-on-return: false
        #打开PSCache，并指定每个连接上PSCache的大小。oracle设为true，mysql设为false。分库分表较多推荐设置为false
        pool-prepared-statements: false
        max-pool-prepared-statement-per-connection-size: 20
        stat:
          merge-sql: true
          log-slow-sql: true
          slow-sql-millis: 2000
            primary: master
      datasource:
        master:
          url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=GMT%2B8
          username: root
          password: root
          driver-class-name: com.mysql.cj.jdbc.Driver
        gbase1:
          url: jdbc:gbase://127.0.0.1:5258/test?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&failOverReadOnly=false&useSSL=false&zeroDateTimeBehavior=convertToNull
          username: gbase
          password: gbase
          driver-class-name: com.gbase.jdbc.Driver
          druid: # 以下参数针对每个库可以重新设置druid参数
            initial-size:
            validation-query: select 1 FROM DUAL #比如oracle就需要重新设置这个
            public-key: #（非全局参数）设置即表示启用加密,底层会自动帮你配置相关的连接参数和filter。

配置好了就可以了，切换数据源的用法和上面的一样的，打@DS(“db1”)注解到service类或方法上就行了
详细配置参考这个配置类com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceProperties

service嵌套

这个就是特性的第九条：提供多层数据源嵌套切换。（ServiceA >>> ServiceB >>> ServiceC，每个Service都是不同的数据源）
借用源码中的demo：实现SchoolService >>> studentService、teacherService

@Service
public class SchoolServiceImpl{
    public void addTeacherAndStudent() {
        teacherService.addTeacherWithTx("ss", 1);
        teacherMapper.addTeacher("test", 111);
        studentService.addStudentWithTx("tt", 2);
    }
}
@Service
@DS("teacher")
public class TeacherServiceImpl {
    public boolean addTeacherWithTx(String name, Integer age) {
        return teacherMapper.addTeacher(name, age);
    }
}
@Service
@DS("student")
public class StudentServiceImpl {
    public boolean addStudentWithTx(String name, Integer age) {
        return studentMapper.addStudent(name, age);
    }
}

这个addTeacherAndStudent调用数据源切换就是primary ->teacher->primary->student->primary

在这里插入图片描述

关于其他demo可以看官方wiki，里面写了很多用法，这里就不赘述了，重点在于学习原理。。。

为什么切换数据源不生效或事务不生效？

这种问题常见于上一节service嵌套，比如serviceA -> serviceB、serviceC，serviceA
加上@Transaction

简单来说：嵌套数据源的service中，如果操作了多个数据源，不能在最外层加上@Transaction开启事务，否则切换数据源不生效，因为这属于分布式事务了，需要用seata方案解决，如果是单个数据源(不需要切换数据源)可以用@Transaction开启事务，保证每个数据源自己的完整性

下面来粗略的分析加事务不生效的原因：
它这个切换数据源的原理就是实现了DataSource接口，实现了getConnection方法，只要在service中开启事务，service中对其他数据源操作只会使用开启事务的数据源，因为开启事务数据源会被缓存下来，可以在DataSourceTransactionManager的doBegin方法中看见那个txObject，如果在一个事务内，就会复用Connection，所以切换不了数据源

/**
	 * This implementation sets the isolation level but ignores the timeout.
	 */
	@Override
	protected void doBegin(Object transaction, TransactionDefinition definition) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		Connection con = null;

		try {
			if (!txObject.hasConnectionHolder() ||
					txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
				// 开启一个新事务会获取一个新的Connection，所以会调用DataSource接口的getConnection方法，从而切换数据源
				Connection newCon = obtainDataSource().getConnection();
				if (logger.isDebugEnabled()) {
					logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
				}
				txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
			}

			txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
			// 如果已经开启了事务，就从holder中获取Connection
			con = txObject.getConnectionHolder().getConnection();
			…………
			}

多数据源事务嵌套
看上面源码，说是新起一个事务才会重新获取Connection，才会成功切换数据源，那我在每个数据源的service方法上都加上@Transaction呢？（涉及spring事务传播行为）

这里做个小实验，还是上面的例子，serviceA ->(嵌套) serviceB、serviceC，serviceA
加上@Transaction，现在给serviceB和serviceC的方法上也加上@Transaction，就是所有service里被调用的方法都打上@Transaction注解

@Transactional
public void addTeacherAndStudentWithTx() {
    teacherService.addTeacherWithTx("ss", 1);
    studentService.addStudentWithTx("tt", 2);
    throw new RuntimeException("test");
}

类似这样，里面两个service也都加上了@Transaction

实际上这样数据源也不会切换，因为默认事务传播级别为required，父子service属于同一事物所以就会用同一Connection。而这里是多数据源，如果把事务传播方式改成require_new给子service起新事物，可以切换数据源，他们都是独立的事务了，然后父service回滚不会导致子service回滚（详见spring事务传播），这样保证了每个单独的数据源的数据完整性，如果要保证所有数据源的完整性，那就用seata分布式事务框架

@Transactional
public void addTeacherAndStudentWithTx() {
	// 做了数据库操作
	aaaDao.doSomethings(“test”);
    teacherService.addTeacherWithTx("ss", 1);
    studentService.addStudentWithTx("tt", 2);
    throw new RuntimeException("test");
}

关于事务嵌套，还有一种情况就是在外部service里面做DB1的一些操作，然后再调用DB2、DB3的service，再想保证DB1的事务，就需要在外部service上加@Transaction，如果想让里面的service正常切换数据源，根据事务传播行为，设置为propagation = Propagation.REQUIRES_NEW就可以了，里面的也能正常切换数据源了，因为它们是独立的事务

补充：关于@Transaction操作多数据源事务的问题

 @Transaction
    public void insertDB1andDB2() {
        db1Service.insertOne();
        db2Service.insertOne();
        throw new RuntimeException("test");
    }

类似于上面这种操作，我们通过注入多个DataSource、DataSourceTransactionManager、SqlSessionFactory、SqlSessionTemplate这四种Bean的方式来实现多数据源(最顶上第一篇博客提到的方式)，然后在外部又加上了@Transaction想实现事务

我试过在中间抛异常查看能不能正常回滚，结果发现只会有一个数据源的事务生效，点开@Transaction注解，发现里面有个transactionManager属性，这个就是指定之前声明的transactionManager Bean，我们默认了DB1的transactionManager为@Primary，所以这时DB2的事务就不会生效，因为用的是DB1的TransactionManager。因为@Transactional只能指定一个事务管理器，并且注解不允许重复，所以就只能使用一个数据源的事务管理器了。如果DB2中的更新失败,我想回滚DB1和DB2以进行回滚，可以使用ChainedTransactionManager来解决，它可以最后尽最大努力回滚事务

源码分析

源码基于3.1.1版本(20200522)
由于篇幅限制，只截了重点代码，如果需要看完整代码可以去github拉，或者点击下载dynamic-datasource-spring-boot-starter.zip

整体结构

在这里插入图片描述

拿到代码要找到入手点，这里带着问题阅读代码

自动配置怎么实现的

一般一个starter的最好入手点就是自动配置类，在 META-INF/spring.factories文件中指定自动配置类入口

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceAutoConfiguration

在spring.factories中看到有这个自动配置
所以从核心自动配置类DynamicDataSourceAutoConfiguration入手
可以认为这就是程序的Main入口

@Slf4j
@Configuration
@AllArgsConstructor
// 以spring.datasource.dynamic为前缀读取配置
@EnableConfigurationProperties(DynamicDataSourceProperties.class)
// 需要在spring boot的DataSource bean自动配置之前注入我们的DataSource bean
@AutoConfigureBefore(DataSourceAutoConfiguration.class)
// 引入了Druid的autoConfig和各种数据源连接池的Creator
@Import(value = {DruidDynamicDataSourceConfiguration.class, DynamicDataSourceCreatorAutoConfiguration.class})
// 当含有spring.datasource.dynamic配置的时候启用这个autoConfig
@ConditionalOnProperty(prefix = DynamicDataSourceProperties.PREFIX, name = "enabled", havingValue = "true", matchIfMissing = true)
public class DynamicDataSourceAutoConfiguration {

    private final DynamicDataSourceProperties properties;

    /**
     * 多数据源加载接口，默认从yml中读取多数据源配置
     * @return DynamicDataSourceProvider
     */
    @Bean
    @ConditionalOnMissingBean
    public DynamicDataSourceProvider dynamicDataSourceProvider() {
        Map<String, DataSourceProperty> datasourceMap = properties.getDatasource();
        return new YmlDynamicDataSourceProvider(datasourceMap);
    }

    /**
     * 注册自己的动态多数据源DataSource
     * @param dynamicDataSourceProvider 各种数据源连接池建造者
     * @return DataSource
     */
    @Bean
    @ConditionalOnMissingBean
    public DataSource dataSource(DynamicDataSourceProvider dynamicDataSourceProvider) {
        DynamicRoutingDataSource dataSource = new DynamicRoutingDataSource();
        dataSource.setPrimary(properties.getPrimary());
        dataSource.setStrict(properties.getStrict());
        dataSource.setStrategy(properties.getStrategy());
        dataSource.setProvider(dynamicDataSourceProvider);
        dataSource.setP6spy(properties.getP6spy());
        dataSource.setSeata(properties.getSeata());
        return dataSource;
    }

    /**
     * AOP切面，对DS注解过的方法进行增强，达到切换数据源的目的
     * @param dsProcessor 动态参数解析数据源，如果数据源名称以#开头，就会进入这个解析器链
     * @return advisor
     */
    @Bean
    @ConditionalOnMissingBean
    public DynamicDataSourceAnnotationAdvisor dynamicDatasourceAnnotationAdvisor(DsProcessor dsProcessor) {
        // aop方法拦截器在方法调用前后做操作
        DynamicDataSourceAnnotationInterceptor interceptor = new DynamicDataSourceAnnotationInterceptor();
        // 动态参数解析器
        interceptor.setDsProcessor(dsProcessor);
        // 使用AbstractPointcutAdvisor将pointcut和advice连接构成切面
        DynamicDataSourceAnnotationAdvisor advisor = new DynamicDataSourceAnnotationAdvisor(interceptor);
        advisor.setOrder(properties.getOrder());
        return advisor;
    }

    /**
     * 动态参数解析器链
     * @return DsProcessor
     */
    @Bean
    @ConditionalOnMissingBean
    public DsProcessor dsProcessor() {
        DsHeaderProcessor headerProcessor = new DsHeaderProcessor();
        DsSessionProcessor sessionProcessor = new DsSessionProcessor();
        DsSpelExpressionProcessor spelExpressionProcessor = new DsSpelExpressionProcessor();
        // 顺序header->session->spel 所有以#开头的参数都会从参数中获取数据源
        headerProcessor.setNextProcessor(sessionProcessor);
        sessionProcessor.setNextProcessor(spelExpressionProcessor);
        return headerProcessor;
    }

    /**
     * 提供不使用注解而使用正则或spel来切换数据源方案（实验性功能）
     * 如果想开启这个功能得自己配置注入DynamicDataSourceConfigure Bean
     * @param dynamicDataSourceConfigure dynamicDataSourceConfigure
     * @param dsProcessor dsProcessor
     * @return advisor
     */
    @Bean
    @ConditionalOnBean(DynamicDataSourceConfigure.class)
    public DynamicDataSourceAdvisor dynamicAdvisor(DynamicDataSourceConfigure dynamicDataSourceConfigure, DsProcessor dsProcessor) {
        DynamicDataSourceAdvisor advisor = new DynamicDataSourceAdvisor(dynamicDataSourceConfigure.getMatchers());
        advisor.setDsProcessor(dsProcessor);
        advisor.setOrder(Ordered.HIGHEST_PRECEDENCE);
        return advisor;
    }
}

这里自动配置的五个Bean都是非常重要的，后面会一一涉及到

这里说说自动配置，主要就是上面自动配置类的几个注解，都写了注释，其中重要的是这个注解：

// 以spring.datasource.dynamic为前缀读取配置
@EnableConfigurationProperties(DynamicDataSourceProperties.class)

@EnableConfigurationProperties：使使用 @ConfigurationProperties 注解的类生效，主要是用来把properties或者yml配置文件转化为bean来使用的，这个在实际使用中非常实用

@ConfigurationProperties(prefix = DynamicDataSourceProperties.PREFIX)
public class DynamicDataSourceProperties {

    public static final String PREFIX = "spring.datasource.dynamic";
    public static final String HEALTH = PREFIX + ".health";

    /**
     * 必须设置默认的库,默认master
     */
    private String primary = "master";
    /**
     * 是否启用严格模式,默认不启动. 严格模式下未匹配到数据源直接报错, 非严格模式下则使用默认数据源primary所设置的数据源
     */
    private Boolean strict = false;
    …………
      /**
     * Druid全局参数配置
     */
    @NestedConfigurationProperty
    private DruidConfig druid = new DruidConfig();
    /**
     * HikariCp全局参数配置
     */
    @NestedConfigurationProperty
    private HikariCpConfig hikari = new HikariCpConfig();
    …………
    }

可以发现之前我们在spring.datasource.dynamic配置的东西都会注入到这个配置Bean中，需要注意的是使用了@NestedConfigurationProperty嵌套了其他的配置类，如果搞不清楚配置项是啥，就直接看看DynamicDataSourceProperties这个类就清楚了

比如说DruidConfig，这个DruidConfig是自定义的一个配置类，不是Druid里面的，它下面有个toProperties方法，为了实现yml配置中每个dataSource下面的durid可以独立配置(不配置就使用全局配置的)，根据全局配置和独立配置结合转换为Properties，然后在DruidDataSourceCreator类中根据这个配置创建druid连接池

如何集成众多连接池的

关于集成连接池配置在上面已经提到过了，就是DynamicDataSourceProperties配置类下，但是如何通过这些配置生成真正的数据源连接池呢，让我们来看creator包

在这里插入图片描述

看名字就知道支持哪几种数据源

在自动配置中，配置DataSource的时候，new了一个DynamicRoutingDataSource，而它实现了InitializingBean接口，在bean初始化时候做一些操作

@Slf4j
public class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean, DisposableBean {
    /**
     * 所有数据库
     */
    private final Map<String, DataSource> dataSourceMap = new LinkedHashMap<>();
    /**
     * 分组数据库
     */
    private final Map<String, DynamicGroupDataSource> groupDataSources = new ConcurrentHashMap<>();
	省略部分代码…………

	/**
     * 添加数据源
     *
     * @param ds         数据源名称
     * @param dataSource 数据源
     */
    public synchronized void addDataSource(String ds, DataSource dataSource) {
        // 如果数据源不存在则保存一个
        if (!dataSourceMap.containsKey(ds)) {
            // 包装seata、p6spy插件
            dataSource = wrapDataSource(ds, dataSource);
            // 保存到所有数据源map
            dataSourceMap.put(ds, dataSource);
            // 对其进行分组并保存map
            this.addGroupDataSource(ds, dataSource);
            log.info("dynamic-datasource - load a datasource named [{}] success", ds);
        } else {
            log.warn("dynamic-datasource - load a datasource named [{}] failed, because it already exist", ds);
        }
    }
    // 包装seata、p6spy插件的方法
    private DataSource wrapDataSource(String ds, DataSource dataSource) {
        if (p6spy) {
            dataSource = new P6DataSource(dataSource);
            log.debug("dynamic-datasource [{}] wrap p6spy plugin", ds);
        }
        if (seata) {
            dataSource = new DataSourceProxy(dataSource);
            log.debug("dynamic-datasource [{}] wrap seata plugin", ds);
        }
        return dataSource;
    }
    // 添加分组数据源的方法
    private void addGroupDataSource(String ds, DataSource dataSource) {
        // 分组用_下划线分割
        if (ds.contains(UNDERLINE)) {
            // 获取组名
            String group = ds.split(UNDERLINE)[0];
            // 如果已存在组，则往里面添加数据源
            if (groupDataSources.containsKey(group)) {
                groupDataSources.get(group).addDatasource(dataSource);
            } else {
                try {
                    // 否则创建一个新的分组
                    DynamicGroupDataSource groupDatasource = new DynamicGroupDataSource(group, strategy.newInstance());
                    groupDatasource.addDatasource(dataSource);
                    groupDataSources.put(group, groupDatasource);
                } catch (Exception e) {
                    log.error("dynamic-datasource - add the datasource named [{}] error", ds, e);
                    dataSourceMap.remove(ds);
                }
            }
        }
    }
   @Override
    public void afterPropertiesSet() throws Exception {
        // 通过配置加载数据源
        Map<String, DataSource> dataSources = provider.loadDataSources();
        // 添加并分组数据源
        for (Map.Entry<String, DataSource> dsItem : dataSources.entrySet()) {
            addDataSource(dsItem.getKey(), dsItem.getValue());
        }
        // 检测默认数据源设置
        if (groupDataSources.containsKey(primary)) {
            log.info("dynamic-datasource initial loaded [{}] datasource,primary group datasource named [{}]", dataSources.size(), primary);
        } else if (dataSourceMap.containsKey(primary)) {
            log.info("dynamic-datasource initial loaded [{}] datasource,primary datasource named [{}]", dataSources.size(), primary);
        } else {
            throw new RuntimeException("dynamic-datasource Please check the setting of primary");
        }
    }
}

这个类就是核心的动态数据源组件，它将DataSource维护在map里，这里重点看如何创建数据源连接池
它所做的操作就是初始化时从provider获取创建好的数据源map，然后解析这个map对其分组，来看看这个provider里面是如何创建这个map的

@Bean
    @ConditionalOnMissingBean
    public DynamicDataSourceProvider dynamicDataSourceProvider() {
        Map<String, DataSourceProperty> datasourceMap = properties.getDatasource();
        return new YmlDynamicDataSourceProvider(datasourceMap);
    }

在自动配置中，注入的是这个bean，就是通过yml读取配置文件的(后面还有通过jdbc读取配置文件)，重点不在这里，这是后面要提到的
通过跟踪provider.loadDataSources();发现在createDataSourceMap方法中调用的是dataSourceCreator.createDataSource(dataSourceProperty)

@Slf4j
@Setter
public class DataSourceCreator {
  	/**
     * 是否存在druid
     */
    private static Boolean druidExists = false;
    /**
     * 是否存在hikari
     */
    private static Boolean hikariExists = false;

    static {
        try {
            Class.forName(DRUID_DATASOURCE);
            druidExists = true;
            log.debug("dynamic-datasource detect druid,Please Notice \n " +
                    "https://github.com/baomidou/dynamic-datasource-spring-boot-starter/wiki/Integration-With-Druid");
        } catch (ClassNotFoundException ignored) {
        }
        try {
            Class.forName(HIKARI_DATASOURCE);
            hikariExists = true;
        } catch (ClassNotFoundException ignored) {
        }
    }
	…………
 	/**
     * 创建数据源
     *
     * @param dataSourceProperty 数据源信息
     * @return 数据源
     */
    public DataSource createDataSource(DataSourceProperty dataSourceProperty) {
        DataSource dataSource;
        //如果是jndi数据源
        String jndiName = dataSourceProperty.getJndiName();
        if (jndiName != null && !jndiName.isEmpty()) {
            dataSource = createJNDIDataSource(jndiName);
        } else {
            Class<? extends DataSource> type = dataSourceProperty.getType();
            // 连接池类型，如果不设置就自动根据Druid > HikariCp的顺序查找
            if (type == null) {
                if (druidExists) {
                    dataSource = createDruidDataSource(dataSourceProperty);
                } else if (hikariExists) {
                    dataSource = createHikariDataSource(dataSourceProperty);
                } else {
                    dataSource = createBasicDataSource(dataSourceProperty);
                }
            } else if (DRUID_DATASOURCE.equals(type.getName())) {
                dataSource = createDruidDataSource(dataSourceProperty);
            } else if (HIKARI_DATASOURCE.equals(type.getName())) {
                dataSource = createHikariDataSource(dataSourceProperty);
            } else {
                dataSource = createBasicDataSource(dataSourceProperty);
            }
        }
        this.runScrip(dataSourceProperty, dataSource);
        return dataSource;
    }
    …………
   }

重点就在这里，根据配置中的type或连接池的class来判断该创建哪种连接池

@Data
@AllArgsConstructor
public class HikariDataSourceCreator {

    private HikariCpConfig hikariCpConfig;

    public DataSource createDataSource(DataSourceProperty dataSourceProperty) {
        HikariConfig config = dataSourceProperty.getHikari().toHikariConfig(hikariCpConfig);
        config.setUsername(dataSourceProperty.getUsername());
        config.setPassword(dataSourceProperty.getPassword());
        config.setJdbcUrl(dataSourceProperty.getUrl());
        config.setDriverClassName(dataSourceProperty.getDriverClassName());
        config.setPoolName(dataSourceProperty.getPoolName());
        return new HikariDataSource(config);
    }
}

比如说创建hikari连接池，就在这个creator中创建了真正的hikari连接池，创建完后放在dataSourceMap维护起来

DS注解如何被拦截处理的

注解拦截处理离不开AOP，所以这里介绍代码中如何使用AOP的

在这里插入图片描述

/**
     * AOP切面，对DS注解过的方法进行增强，达到切换数据源的目的
     * @param dsProcessor 动态参数解析数据源，如果数据源名称以#开头，就会进入这个解析器链
     * @return advisor
     */
    @Bean
    @ConditionalOnMissingBean
    public DynamicDataSourceAnnotationAdvisor dynamicDatasourceAnnotationAdvisor(DsProcessor dsProcessor) {
        // aop方法拦截器在方法调用前后做操作
        DynamicDataSourceAnnotationInterceptor interceptor = new DynamicDataSourceAnnotationInterceptor();
        // 动态参数解析器
        interceptor.setDsProcessor(dsProcessor);
        // 使用AbstractPointcutAdvisor将pointcut和advice连接构成切面
        DynamicDataSourceAnnotationAdvisor advisor = new DynamicDataSourceAnnotationAdvisor(interceptor);
        advisor.setOrder(properties.getOrder());
        return advisor;
    }

    /**
     * 动态参数解析器链
     * @return DsProcessor
     */
    @Bean
    @ConditionalOnMissingBean
    public DsProcessor dsProcessor() {
        DsHeaderProcessor headerProcessor = new DsHeaderProcessor();
        DsSessionProcessor sessionProcessor = new DsSessionProcessor();
        DsSpelExpressionProcessor spelExpressionProcessor = new DsSpelExpressionProcessor();
        // 顺序header->session->spel 所有以#开头的参数都会从参数中获取数据源
        headerProcessor.setNextProcessor(sessionProcessor);
        sessionProcessor.setNextProcessor(spelExpressionProcessor);
        return headerProcessor;
    }

还是从这个自动配置类入手，发现注入了一个DynamicDataSourceAnnotationAdvisor bean，它是一个advisor

阅读这个advisor之前，这里多提一点AOP相关的

在 Spring AOP 中，有 3 个常用的概念，Advices 、 Pointcut 、 Advisor ，解释如下：
Advices ：表示一个 method 执行前或执行后的动作。
Pointcut ：表示根据 method 的名字或者正则表达式等方式去拦截一个 method 。
Advisor ： Advice 和 Pointcut 组成的独立的单元，并且能够传给 proxy factory 对象。

@Component
//声明这是一个切面Bean
@Aspect
public class ServiceAspect {
    //配置切入点,该方法无方法体,主要为方便同类中其他方法使用此处配置的切入点
    @Pointcut("execution(* com.xxx.aop.service..*(..))")
    public void aspect() {
    }

    /*
     * 配置前置通知,使用在方法aspect()上注册的切入点
     * 同时接受JoinPoint切入点对象,可以没有该参数
     */
    @Before("aspect()")
    public void before(JoinPoint joinPoint) {
    }

    //配置后置通知,使用在方法aspect()上注册的切入点
    @After("aspect()")
    public void after(JoinPoint joinPoint) {
    }

    //配置环绕通知,使用在方法aspect()上注册的切入点
    @Around("aspect()")
    public void around(JoinPoint joinPoint) {
    }

    //配置后置返回通知,使用在方法aspect()上注册的切入点
    @AfterReturning("aspect()")
    public void afterReturn(JoinPoint joinPoint) {
    }

    //配置抛出异常后通知,使用在方法aspect()上注册的切入点
    @AfterThrowing(pointcut = "aspect()", throwing = "ex")
    public void afterThrow(JoinPoint joinPoint, Exception ex) {
    }
}

我们平常可能使用这种AspectJ注解多一点，通过@Aspect注解的方式来声明切面，spring会通过我们的AspectJ注解（比如@Pointcut、@Before）动态的生成各个Advisor。

Spring还提供了另一种切面-顾问（Advisor），其可以完成更为复杂的切面织入功能，我们可以通过直接继承AbstractPointcutAdvisor来提供切面逻辑。
它们最终都会生成对应的Advisor实例

而这里就是使用了继承AbstractPointcutAdvisor的方式来实现切面的

在这里插入图片描述

其中最重要的就是getAdvice和getPointcut方法，可以简单的认为advisor=advice+pointcut

public class DynamicDataSourceAnnotationAdvisor extends AbstractPointcutAdvisor implements
        BeanFactoryAware {

    // 通知
    private Advice advice;

    // 切入点
    private Pointcut pointcut;

    public DynamicDataSourceAnnotationAdvisor(@NonNull DynamicDataSourceAnnotationInterceptor dynamicDataSourceAnnotationInterceptor) {
        this.advice = dynamicDataSourceAnnotationInterceptor;
        this.pointcut = buildPointcut();
    }

    @Override
    public Pointcut getPointcut() {
        return this.pointcut;
    }

    @Override
    public Advice getAdvice() {
        return this.advice;
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        if (this.advice instanceof BeanFactoryAware) {
            ((BeanFactoryAware) this.advice).setBeanFactory(beanFactory);
        }
    }

    private Pointcut buildPointcut() {
        //类级别
        Pointcut cpc = new AnnotationMatchingPointcut(DS.class, true);
        //方法级别
        Pointcut mpc = AnnotationMatchingPointcut.forMethodAnnotation(DS.class);
        //对于类和方法上都可以添加注解的情况
        //类上的注解,最终会将注解绑定到每个方法上
        return new ComposablePointcut(cpc).union(mpc);
    }
}

现在再来看@DS注解的advisor实现，在buildPointcut方法里拦截了被@DS注解的方法或类，并且使用ComposablePointcut组合切入点，可以实现方法优先级大于类优先级的特性
发现advice是通过构造方法传来的，是DynamicDataSourceAnnotationInterceptor，现在来看看这个

public class DynamicDataSourceAnnotationInterceptor implements MethodInterceptor {

    /**
     * The identification of SPEL.
     */
    private static final String DYNAMIC_PREFIX = "#";
    private static final DataSourceClassResolver RESOLVER = new DataSourceClassResolver();
    @Setter
    private DsProcessor dsProcessor;

    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
        try {
            // 这里把获取到的数据源标识如master存入本地线程
            DynamicDataSourceContextHolder.push(determineDatasource(invocation));
            return invocation.proceed();
        } finally {
            DynamicDataSourceContextHolder.poll();
        }
    }

    private String determineDatasource(MethodInvocation invocation) throws Throwable {
        //获得DS注解的方法
        Method method = invocation.getMethod();
        DS ds = method.isAnnotationPresent(DS.class) ? method.getAnnotation(DS.class)
                : AnnotationUtils.findAnnotation(RESOLVER.targetClass(invocation), DS.class);
        //获得DS注解的内容
        String key = ds.value();
        //如果DS注解内容是以#开头解析动态最终值否则直接返回
        return (!key.isEmpty() && key.startsWith(DYNAMIC_PREFIX)) ? dsProcessor.determineDatasource(invocation, key) : key;
    }
}

这是它的advice通知，也可以说是方法拦截器，在要切换数据源的方法前，将切换的数据源放入了holder里，方法执行完后在finally中释放掉，也就是在这里做了当前数据源的切换。下面的determineDatasource决定数据源的方法中判断了以#开头解析动态参数数据源，这个功能就是特性中说的使用spel动态参数解析数据源，如从session，header或参数中获取数据源。

剩下的还有个DynamicDataSourceAdvisor，这个功能是特性八的提供不使用注解而使用正则或spel来切换数据源方案（实验性功能），这里就不介绍这块了

多数据源动态切换及如何管理多数据源

在上一节AOP实现里面的MethodInterceptor里，在方法前后调用了DynamicDataSourceContextHolder.push()和poll()，这个holder类似于前一篇博客使用AbstractRoutingDataSource做多数据源动态切换用的holder，只是这里做了点改造

public final class DynamicDataSourceContextHolder {

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

    private DynamicDataSourceContextHolder() {
    }

    /**
     * 获得当前线程数据源
     *
     * @return 数据源名称
     */
    public static String peek() {
        return LOOKUP_KEY_HOLDER.get().peek();
    }

    /**
     * 设置当前线程数据源
     * <p>
     * 如非必要不要手动调用，调用后确保最终清除
     * </p>
     *
     * @param ds 数据源名称
     */
    public static void push(String ds) {
        LOOKUP_KEY_HOLDER.get().push(StringUtils.isEmpty(ds) ? "" : ds);
    }

    /**
     * 清空当前线程数据源
     * <p>
     * 如果当前线程是连续切换数据源 只会移除掉当前线程的数据源名称
     * </p>
     */
    public static void poll() {
        Deque<String> deque = LOOKUP_KEY_HOLDER.get();
        deque.poll();
        if (deque.isEmpty()) {
            LOOKUP_KEY_HOLDER.remove();
        }
    }

    /**
     * 强制清空本地线程
     * <p>
     * 防止内存泄漏，如手动调用了push可调用此方法确保清除
     * </p>
     */
    public static void clear() {
        LOOKUP_KEY_HOLDER.remove();
    }
}

它使用了栈这个数据结构当前数据源，使用了ArrayDeque这个线程不安全的双端队列容器来实现栈功能，它作为栈性能比Stack好，现在不推荐用老容器
用栈的话，嵌套过程中push，出去就pop，实现了这个嵌套调用service的业务需求

现在来看切换数据源的核心类

在这里插入图片描述

在之前做动态数据源切换的时候，我们利用Spring的AbstractRoutingDataSource做多数据源动态切换，它实现了DataSource接口，重写了getConnection方法
在这里切换数据源原理也是如此，它自己写了一个AbstractRoutingDataSource类，不是spring的那个，现在来看看这个类

public abstract class AbstractRoutingDataSource extends AbstractDataSource {

    /**
     * 子类实现决定最终数据源
     *
     * @return 数据源
     */
    protected abstract DataSource determineDataSource();

    @Override
    public Connection getConnection() throws SQLException {
        return determineDataSource().getConnection();
    }

    @Override
    public Connection getConnection(String username, String password) throws SQLException {
        return determineDataSource().getConnection(username, password);
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T unwrap(Class<T> iface) throws SQLException {
        if (iface.isInstance(this)) {
            return (T) this;
        }
        return determineDataSource().unwrap(iface);
    }

    @Override
    public boolean isWrapperFor(Class<?> iface) throws SQLException {
        return (iface.isInstance(this) || determineDataSource().isWrapperFor(iface));
    }
}

可以发现也是实现了DataSource接口的getConnection方法，现在来看下子类如何实现determineDataSource方法的

public class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean, DisposableBean {

    private static final String UNDERLINE = "_";
    /**
     * 所有数据库
     */
    private final Map<String, DataSource> dataSourceMap = new LinkedHashMap<>();
    /**
     * 分组数据库
     */
    private final Map<String, DynamicGroupDataSource> groupDataSources = new ConcurrentHashMap<>();
    }
    
    @Override
    public DataSource determineDataSource() {
        return getDataSource(DynamicDataSourceContextHolder.peek());
    }

    private DataSource determinePrimaryDataSource() {
        log.debug("dynamic-datasource switch to the primary datasource");
        return groupDataSources.containsKey(primary) ? groupDataSources.get(primary).determineDataSource() : dataSourceMap.get(primary);
    }
     /**
     * 获取数据源
     *
     * @param ds 数据源名称
     * @return 数据源
     */
    public DataSource getDataSource(String ds) {
        if (StringUtils.isEmpty(ds)) {
            return determinePrimaryDataSource();
        } else if (!groupDataSources.isEmpty() && groupDataSources.containsKey(ds)) {
            log.debug("dynamic-datasource switch to the datasource named [{}]", ds);
            return groupDataSources.get(ds).determineDataSource();
        } else if (dataSourceMap.containsKey(ds)) {
            log.debug("dynamic-datasource switch to the datasource named [{}]", ds);
            return dataSourceMap.get(ds);
        }
        if (strict) {
            throw new RuntimeException("dynamic-datasource could not find a datasource named" + ds);
        }
        return determinePrimaryDataSource();
    }
    …………
    }

之前creator生成的数据源连接池放入map维护后，现在获取数据源就是从map中取就行了，可以发现这里数据组优先于单数据源

数据组的负载均衡怎么做的

在上一节中，DynamicRoutingDataSource的getDataSource方法里

else if (!groupDataSources.isEmpty() && groupDataSources.containsKey(ds)) {
            log.debug("dynamic-datasource switch to the datasource named [{}]", ds);
            return groupDataSources.get(ds).determineDataSource();
        }

如果数据组不为空并且DS注解写的数据组名，那么就会在数据组中选取一个数据源，调用的determineDataSource方法

@Data
public class DynamicGroupDataSource {

    private String groupName;

    // 数据源切换策略
    private DynamicDataSourceStrategy dynamicDataSourceStrategy;

    private List<DataSource> dataSources = new LinkedList<>();

    public DynamicGroupDataSource(String groupName, DynamicDataSourceStrategy dynamicDataSourceStrategy) {
        this.groupName = groupName;
        this.dynamicDataSourceStrategy = dynamicDataSourceStrategy;
    }

    public void addDatasource(DataSource dataSource) {
        dataSources.add(dataSource);
    }

    public void removeDatasource(DataSource dataSource) {
        dataSources.remove(dataSource);
    }

    // 根据切换策略，决定一个数据源
    public DataSource determineDataSource() {
        return dynamicDataSourceStrategy.determineDataSource(dataSources);
    }

    public int size() {
        return dataSources.size();
    }
}

这是数据组的DataSource，里面根据策略模式来决定一个数据源，目前实现的就两种，随机和轮询，默认的是轮询，在DynamicDataSourceProperties属性中写了默认值，也可以通过配置文件配置

public class LoadBalanceDynamicDataSourceStrategy implements DynamicDataSourceStrategy {

    /**
     * 负载均衡计数器
     */
    private final AtomicInteger index = new AtomicInteger(0);

    @Override
    public DataSource determineDataSource(List<DataSource> dataSources) {
        return dataSources.get(Math.abs(index.getAndAdd(1) % dataSources.size()));
    }
}

这是一个简单的轮询负载均衡，我们可以通过自己的业务需求，新增一个策略类来实现新的负载均衡算法

如何自定义数据配置来源

默认是从yml中读取数据源配置的(YmlDynamicDataSourceProvider)，实际业务中，我们可能遇到从其他地方获取配置来创建数据源，比如从数据库、配置中心、mq等等

在这里插入图片描述

想自定义数据来源可以自定义一个provider实现DynamicDataSourceProvider接口并继承AbstractDataSourceProvider类就行了

public interface DynamicDataSourceProvider {
    /**
     * 加载所有数据源
     *
     * @return 所有数据源，key为数据源名称
     */
    Map<String, DataSource> loadDataSources();
}

如果想通过jdbc获取数据源，它这里有个抽象类AbstractJdbcDataSourceProvider，需要实现它的executeStmt方法，就是从其他数据库查询出这些信息，url、username、password等等(就是我们在yml配置的那些信息)，然后拼接成一个配置对象DataSourceProperty返回出去调用createDataSourceMap方法就行了

如何动态增减数据源

这个也是实际中很实用的功能，它的实现还是通过DynamicRoutingDataSource这个核心动态数据源组件来做的

@Slf4j
public class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean, DisposableBean {
   /**
     * 所有数据库
     */
    private final Map<String, DataSource> dataSourceMap = new LinkedHashMap<>();
    /**
     * 分组数据库
     */
    private final Map<String, DynamicGroupDataSource> groupDataSources = new ConcurrentHashMap<>();
    …………
      /**
     * 获取当前所有的数据源
     *
     * @return 当前所有数据源
     */
    public Map<String, DataSource> getCurrentDataSources() {
        return dataSourceMap;
    }

    /**
     * 获取的当前所有的分组数据源
     *
     * @return 当前所有的分组数据源
     */
    public Map<String, DynamicGroupDataSource> getCurrentGroupDataSources() {
        return groupDataSources;
    }
      /**
     * 添加数据源
     *
     * @param ds         数据源名称
     * @param dataSource 数据源
     */
    public synchronized void addDataSource(String ds, DataSource dataSource) {
        // 如果数据源不存在则保存一个
        if (!dataSourceMap.containsKey(ds)) {
            // 包装seata、p6spy插件
            dataSource = wrapDataSource(ds, dataSource);
            // 保存
            dataSourceMap.put(ds, dataSource);
            // 对其进行分组
            this.addGroupDataSource(ds, dataSource);
            log.info("dynamic-datasource - load a datasource named [{}] success", ds);
        } else {
            log.warn("dynamic-datasource - load a datasource named [{}] failed, because it already exist", ds);
        }
    }
    /**
     * 删除数据源
     *
     * @param ds 数据源名称
     */
    public synchronized void removeDataSource(String ds) {
        if (!StringUtils.hasText(ds)) {
            throw new RuntimeException("remove parameter could not be empty");
        }
        if (primary.equals(ds)) {
            throw new RuntimeException("could not remove primary datasource");
        }
        if (dataSourceMap.containsKey(ds)) {
            DataSource dataSource = dataSourceMap.get(ds);
            try {
                closeDataSource(ds, dataSource);
            } catch (Exception e) {
                throw new RuntimeException("dynamic-datasource - remove the database named " + ds + " failed", e);
            }
            dataSourceMap.remove(ds);
            if (ds.contains(UNDERLINE)) {
                String group = ds.split(UNDERLINE)[0];
                if (groupDataSources.containsKey(group)) {
                    groupDataSources.get(group).removeDatasource(dataSource);
                }
            }
            log.info("dynamic-datasource - remove the database named [{}] success", ds);
        } else {
            log.warn("dynamic-datasource - could not find a database named [{}]", ds);
        }
    }
    …………
}

可以发现它预留了相关接口给开发者，可方便的添加删除数据库

添加数据源我们需要做的就是：
1、注入DynamicRoutingDataSource和DataSourceCreator
2、通过数据源配置(url、username、password等)构建一个DataSourceProperty对象
3、再通过dataSourceCreator根据配置构建一个真实的DataSource
4、最后调用DynamicRoutingDataSource的addDataSource方法添加这个DataSource就行了
同理，删除数据源：
1、注入DynamicRoutingDataSource
2、调用DynamicRoutingDataSource的removeDataSource方法

  @PostMapping("/add")
    @ApiOperation("通用添加数据源（推荐）")
    public Set<String> add(@Validated @RequestBody DataSourceDTO dto) {
        DataSourceProperty dataSourceProperty = new DataSourceProperty();
        BeanUtils.copyProperties(dto, dataSourceProperty);
        DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
        DataSource dataSource = dataSourceCreator.createDataSource(dataSourceProperty);
        ds.addDataSource(dto.getPollName(), dataSource);
        return ds.getCurrentDataSources().keySet();
    }
    @DeleteMapping
    @ApiOperation("删除数据源")
    public String remove(String name) {
        DynamicRoutingDataSource ds = (DynamicRoutingDataSource) dataSource;
        ds.removeDataSource(name);
        return "删除成功";
    }

总结

通过阅读这块源码，涉及到了一些spring aop、spring事务管理、spring boot自动配置等等，可以更加熟悉使用spring的这些扩展点、api等，还可以根据业务需求去扩展这个starter

加载全部内容

springboot starter多数据源 使用dynamic datasource springboot starter实现多数据源及源码分析

简介

实操

基本使用

集成druid连接池

service嵌套

为什么切换数据源不生效或事务不生效？

源码分析

整体结构

自动配置怎么实现的

如何集成众多连接池的

DS注解如何被拦截处理的

多数据源动态切换及如何管理多数据源

数据组的负载均衡怎么做的

如何自定义数据配置来源

如何动态增减数据源

总结

springboot starter多数据源使用dynamic datasource springboot starter实现多数据源及源码分析