使用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
到此这篇关于使用dynamic-datasource-spring-boot-starter做多数据源及源码分析的文章就介绍到这了,更多相关dynamic-datasource-spring-boot-starter多数据源内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！