使用dynamic-datasource-spring-boot-starter做多数据源及源码分析
多数据源系列
1、spring boot2.0 +Mybatis + druid搭建一个最简单的多数据源
2、利用Spring的AbstractRoutingDataSource做多数据源动态切换
3、使用dynamic-datasource-spring-boot-starter做多数据源及源码分析
文章目录
多数据源系列
简介
实操
基本使用
集成druid连接池
service嵌套
为什么切换数据源不生效或事务不生效？
源码分析
整体结构
自动配置怎么实现的
如何集成众多连接池的
DS注解如何被拦截处理的
多数据源动态切换及如何管理多数据源
数据组的负载均衡怎么做的
如何自定义数据配置来源
如何动态增减数据源
总结
简介
前两篇博客介绍了用基本的方式做多数据源，可以应对一般的情况，但是遇到一些复杂的情况就需要扩展下功能了，比如：动态增减数据源、数据源分组，纯粹多库 读写分离 一主多从、从其他数据库或者配置中心读取数据源等等。其实就算没有这些需求，使用这个实现多数据源也比之前使用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
1
2
在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)
1
2
@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 "删除成功";
}