如何使用dynamic-datasource-spring-boot-starter实现多数据源及源码分析
本篇文章给大家分享的是有关如何使用dynamic-datasource-spring-boot-starter实现多数据源及源码分析，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
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的分布式事务支持。
实操
先把坐标丢出来
com.baomidou dynamic-datasource-spring-boot-starter 3.1.0
下面抽几个用的比较多的应用场景介绍
基本使用
使用方法很简洁，分两步走
一：通过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> selectAll() { return jdbcTemplate.queryForList("select * from user"); } @Override @DS("db1") public List> selectByCondition() { return jdbcTemplate.queryForList("select * from user where age >10"); }} | 注解 | 结果 |
|---|---|
| 没有@DS | 默认数据源 |
| @DS("dsName") | dsName可以为组名也可以为具体某个库的名称 |
通过日志可以发现我们配置的多数据源已经被初始化了，如果切换数据源也会看到打印日子的
是不是很便捷，这是官方的例子
集成druid连接池
com.alibaba druid-spring-boot-starter 1.1.22
首先引入依赖
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
@Servicepublic 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注解
@Transactionalpublic 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分布式事务框架
@Transactionalpublic 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 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初始化时候做一些操作
@Slf4jpublic class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean, DisposableBean { /** * 所有数据库 */ private final Map dataSourceMap = new LinkedHashMap<>(); /** * 分组数据库 */ private final Map 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 dataSources = provider.loadDataSources(); // 添加并分组数据源 for (Map.Entry 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 datasourceMap = properties.getDatasource(); return new YmlDynamicDataSourceProvider(datasourceMap); } 在自动配置中，注入的是这个bean，就是通过yml读取配置文件的(后面还有通过jdbc读取配置文件)，重点不在这里，这是后面要提到的
通过跟踪provider.loadDataSources();发现在createDataSourceMap方法中调用的是dataSourceCreator.createDataSource(dataSourceProperty)
@Slf4j@Setterpublic 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 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@AllArgsConstructorpublic 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@Aspectpublic 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 { /** * 为什么要用链表存储(准确的是栈) * * 为了支持嵌套切换，如ABC三个service都是不同的数据源 * 其中A的某个业务要调B的方法，B的方法需要调用C的方法。一级一级调用切换，形成了链。 * 传统的只设置当前线程的方式不能满足此业务需求，必须使用栈，后进先出。 *
*/ private static final ThreadLocal> LOOKUP_KEY_HOLDER = new NamedThreadLocal>("dynamic-datasource") { @Override protected Deque initialValue() { return new ArrayDeque<>(); } }; private DynamicDataSourceContextHolder() { } /** * 获得当前线程数据源 * * @return 数据源名称 */ public static String peek() { return LOOKUP_KEY_HOLDER.get().peek(); } /** * 设置当前线程数据源 * * 如非必要不要手动调用，调用后确保最终清除 *
* * @param ds 数据源名称 */ public static void push(String ds) { LOOKUP_KEY_HOLDER.get().push(StringUtils.isEmpty(ds) ? "" : ds); } /** * 清空当前线程数据源 * * 如果当前线程是连续切换数据源 只会移除掉当前线程的数据源名称 *
*/ public static void poll() { Deque deque = LOOKUP_KEY_HOLDER.get(); deque.poll(); if (deque.isEmpty()) { LOOKUP_KEY_HOLDER.remove(); } } /** * 强制清空本地线程 * * 防止内存泄漏，如手动调用了push可调用此方法确保清除 *
*/ 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 unwrap(Class 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 dataSourceMap = new LinkedHashMap<>(); /** * 分组数据库 */ private final Map 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方法
@Datapublic class DynamicGroupDataSource { private String groupName; // 数据源切换策略 private DynamicDataSourceStrategy dynamicDataSourceStrategy; private List 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 dataSources) { return dataSources.get(Math.abs(index.getAndAdd(1) % dataSources.size())); }} 这是一个简单的轮询负载均衡，我们可以通过自己的业务需求，新增一个策略类来实现新的负载均衡算法
如何自定义数据配置来源
默认是从yml中读取数据源配置的(YmlDynamicDataSourceProvider)，实际业务中，我们可能遇到从其他地方获取配置来创建数据源，比如从数据库、配置中心、mq等等
想自定义数据来源可以自定义一个provider实现DynamicDataSourceProvider接口并继承AbstractDataSourceProvider类就行了
public interface DynamicDataSourceProvider { /** * 加载所有数据源 * * @return 所有数据源，key为数据源名称 */ Map loadDataSources();} 如果想通过jdbc获取数据源，它这里有个抽象类AbstractJdbcDataSourceProvider，需要实现它的executeStmt方法，就是从其他数据库查询出这些信息，url、username、password等等(就是我们在yml配置的那些信息)，然后拼接成一个配置对象DataSourceProperty返回出去调用createDataSourceMap方法就行了
如何动态增减数据源
这个也是实际中很实用的功能，它的实现还是通过DynamicRoutingDataSource这个核心动态数据源组件来做的
@Slf4jpublic class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean, DisposableBean { /** * 所有数据库 */ private final Map dataSourceMap = new LinkedHashMap<>(); /** * 分组数据库 */ private final Map groupDataSources = new ConcurrentHashMap<>(); ………… /** * 获取当前所有的数据源 * * @return 当前所有数据源 */ public Map getCurrentDataSources() { return dataSourceMap; } /** * 获取的当前所有的分组数据源 * * @return 当前所有的分组数据源 */ public Map 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 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。
以上就是如何使用dynamic-datasource-spring-boot-starter实现多数据源及源码分析，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注行业资讯频道。
