dynamic-datasource-spring-boot-starter读写分离分析
李弘光
地址
简介
主要的功能有：
- 数据源分组，适用于多种场景 纯粹多库 读写分离 一主多从 混合模式。
- 内置敏感参数加密和启动初始化表结构schema数据库database。
- 提供对Druid，Mybatis-Plus，P6sy，Jndi的快速集成。
- 简化Druid和HikariCp配置，提供全局参数配置。
- 提供自定义数据源来源接口(默认使用yml或properties配置)。
- 提供项目启动后增减数据源方案。
- 提供Mybatis环境下的 纯读写分离 方案。
- 使用spel动态参数解析数据源，如从session，header或参数中获取数据源。（多租户架构神器）
- 提供多层数据源嵌套切换。（ServiceA >>> ServiceB >>> ServiceC，每个Service都是不同的数据源）
- 提供 不使用注解 而 使用 正则 或 spel 来切换数据源方案（实验性功能）。
国内大部分的项目，数据库的orm大多数都用的是mybatis。而使用mybatis
让人很纠结的一点就是“自动生成代码”，不自动生成代码吧，要自己手写很多代码；自动生成代码吧，当表结构发生变更的时候又很尴尬。在这种情况下，很多人就会使用mybatis-plus来避免这些问题。而dynamic-datasource-spring-boot-starter是属于mybatis-plus的生态圈， 根据官方文档，是多mybatis-plus2.x和3.x做了专门适配，有天然的优势。
代码分析
研究一个starter的源码，最好的入手点就是MATE-INF/spring.factories。从这里可以看到自动配置类，在通过自动配置类，就可以知道代码是怎么起作用的。
spring.factories
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceAutoConfiguration
可以看到，只有一个自动配置类，DynamicDataSourceAutoConfiguration
DynamicDataSourceAutoConfiguration
@Slf4j
@Configuration
@EnableConfigurationProperties(DynamicDataSourceProperties.class)
// 在spring-jdbc的自动配置之前先定义好datasource
@AutoConfigureBefore(DataSourceAutoConfiguration.class)
// 引入druid的自动配置类
@Import(DruidDynamicDataSourceConfiguration.class)
@ConditionalOnProperty(prefix = DynamicDataSourceProperties.PREFIX, name="enabled", havingValue="true", matchIfMissing = true)
public class DynamicDataSourceAutoConfiguration {
@Autowired
private DynamicDataSourceProperties properties;
// 用于生成一个“库名->数据源”的map
@Bean
@ConditionalOnMissingBean
public DynamicDataSourceProvider dynamicDataSourceProvider() {
return new YmlDynamicDataSourceProvider(properties);
}
// 用户生成datasource，可以根据引入的包，自动识别要创建的datasource类型
// DynamicDataSourceProvider就是使用DynamicDataSourceCreator来创建datasurec，放到map中
@Bean
@ConditionalOnMissingBean
public DynamicDataSourceCreator dynamicDataSourceCreator() {
DynamicDataSourceCreator dynamicDataSourceCreator = new DynamicDataSourceCreator();
dynamicDataSourceCreator.setDruidGlobalConfig(properties.getDruid());
dynamicDataSourceCreator.setHikariGlobalConfig(properties.getHikari());
dynamicDataSourceCreator.setGlobalPublicKey(properties.getPublicKey());
return dynamicDataSourceCreator;
}
// 定义DataSource的Bean，DynamicRoutingDataSource利用DynamicDataSourceProvider生成了“库名->数据源”的map
// 在获取真实数据源的时候，再根据ThreadLocal里的变量，决定选取map中的那个datasource
// 注意properties.getPrimary()，这个默认是master，即默认走主库
@Bean
@ConditionalOnMissingBean
public DataSource dataSource(DynamicDataSourceProvider dynamicDataSourceProvider) {
DynamicRoutingDataSource dataSource = new DynamicRoutingDataSource();
dataSource.setPrimary(properties.getPrimary());
dataSource.setStrategy(properties.getStrategy());
dataSource.setProvider(dynamicDataSourceProvider);
dataSource.setP6spy(properties.getP6spy());
dataSource.setStrict(properties.getStrict());
return dataSource;
}
// @Ds注解的advisor，只要在类或者方法上，增加了@Ds注解，就会被拦截：在方法执行前根据@Ds的value，往ThreadLocal设置要访问的数据源；在方法执行结束后，清除ThreadLocal中的值。
@Bean
@ConditionalOnMissingBean
public DynamicDataSourceAnnotationAdvisor dynamicDatasourceAnnotationAdvisor(
DsProcessor dsProcessor) {
DynamicDataSourceAnnotationInterceptor interceptor = new DynamicDataSourceAnnotationInterceptor();
interceptor.setDsProcessor(dsProcessor);
DynamicDataSourceAnnotationAdvisor advisor = new DynamicDataSourceAnnotationAdvisor(
interceptor);
advisor.setOrder(properties.getOrder());
return advisor;
}
// 定义一个决定目标数据源的责任链，先从http的header获取，获取不到再从http的session获取，还获取不到，就会通过spel表达式来获取
// 感觉这个有点奇怪啊，为什么要做这样的功能，由请求者指定要从哪个库读数据？？？请求者还得知道服务有哪些数据库，叫什么名字哦？？？
@Bean
@ConditionalOnMissingBean
public DsProcessor dsProcessor() {
DsHeaderProcessor headerProcessor = new DsHeaderProcessor();
DsSessionProcessor sessionProcessor = new DsSessionProcessor();
DsSpelExpressionProcessor spelExpressionProcessor = new DsSpelExpressionProcessor();
headerProcessor.setNextProcessor(sessionProcessor);
sessionProcessor.setNextProcessor(spelExpressionProcessor);
return headerProcessor;
}
// 定义基于表达式的advisor，对符合条件的点用DsProcessor来判断要访问的数据库
@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的作用。只是单纯要了解读写分离的原理，不用去关心DynamicDataSourceProvider、DynamicDataSourceCreator这两个的代码。只需要关注DynamicRoutingDataSource、DynamicDataSourceAnnotationAdvisor、DsProcessor 、DynamicDataSourceAdvisor这四个类的代码。
DynamicRoutingDataSource
@Slf4j
public class DynamicRoutingDataSource extends AbstractRoutingDataSource implements InitializingBean,
DisposableBean {
// 数据源名称包含下划线，则会把下划线分割的第一部分作为数据源组名
private static final String UNDERLINE = "_";
// 用于生成真实数据源的map
@Setter
private DynamicDataSourceProvider provider;
// 当同一个分组有多个数据源时，采用的负载均衡算法，目前支持轮询和随机访问两种，分别是LoadBalanceDynamicDataSourceStrategy和RandomDynamicDataSourceStrategy类
@Setter
private Class<? extends DynamicDataSourceStrategy> strategy;
// 默认数据源的名称或分组名称
@Setter
private String primary;
// 是否保持粘性，即访问了某个数据源，接下来就一直访问那个数据源
@Setter
private boolean strict;
// 不知道是啥，也不关系，跟读写分离的内容关系不大
private boolean p6spy;
/**
* 所有数据库
*/
private Map<String, DataSource> dataSourceMap = new LinkedHashMap<>();
/**
* 分组数据库
*/
private Map<String, DynamicGroupDataSource> groupDataSources = new ConcurrentHashMap<>();
// 该类继承了spring的AbstractRoutingDataSource类，所以需要实现它的抽象方法，选择数据源。这里需要关注的是DynamicDataSourceContextHolder类
@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);
}
/**
* 获取当前所有的数据源
*
* @return 当前所有数据源
*/
public Map<String, DataSource> getCurrentDataSources() {
return dataSourceMap;
}
/**
* 获取的当前所有的分组数据源
*
* @return 当前所有的分组数据源
*/
public Map<String, DynamicGroupDataSource> getCurrentGroupDataSources() {
return groupDataSources;
}
/**
* 获取数据源
*
* @param ds 数据源名称
* @return 数据源
*/
public DataSource getDataSource(String ds) {
if (StringUtils.isEmpty(ds)) {
// 如果没有指定数据源，则调用determinePrimaryDataSource方法来选择默认数据源
return determinePrimaryDataSource();
} else if (!groupDataSources.isEmpty() && groupDataSources.containsKey(ds)) {
// 如果有数据源分组，则判断指定的是否是数据源分组，例如配置了数据源slave_1,slave_2，这个时候，指定使用的数据源是slave，则会先选择slave分组，再从分组里选择一个数据源
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();
}
// addDataSource和removeDataSource两个方法，就是实现动态增删数据源的基础
/**
* 添加数据源
*
* @param ds 数据源名称
* @param dataSource 数据源
*/
public synchronized void addDataSource(String ds, DataSource dataSource) {
if (p6spy) {
dataSource = new P6DataSource(dataSource);
}
dataSourceMap.put(ds, dataSource);
// 数据源名称包含下划线，就获取分组名称，设置到DynamicGroupDataSource中去
// 这个感觉不是很多，依赖下划线来实现分组，个人觉得，应该用层级来实现，例如master.m1、master.m2来代替master_m1、master_m2
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);
}
}
}
log.info("dynamic-datasource - load a datasource named [{}] success", ds);
}
/**
* 删除数据源
*
* @param ds 数据源名称
*/
public synchronized void removeDataSource(String ds) {
.......
}
// 销毁bean的时候，需要调用所有真实数据源的close方法，关闭数据源
@Override
public void destroy() throws Exception {
log.info("dynamic-datasource start closing ....");
for (Map.Entry<String, DataSource> item : dataSourceMap.entrySet()) {
DataSource dataSource = item.getValue();
if (p6spy) {
Field realDataSourceField = P6DataSource.class.getDeclaredField("realDataSource");
realDataSourceField.setAccessible(true);
dataSource = (DataSource) realDataSourceField.get(dataSource);
}
Class<? extends DataSource> clazz = dataSource.getClass();
try {
Method closeMethod = clazz.getDeclaredMethod("close");
closeMethod.invoke(dataSource);
} catch (NoSuchMethodException e) {
log.warn("dynamic-datasource close the datasource named [{}] failed,", item.getKey());
}
}
log.info("dynamic-datasource all closed success,bye");
}
// 完成初始化之后，需要对默认数据源做一个校验，如果不包含默认数据源，则直接报错
@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");
}
}
}
看了DynamicRoutingDataSource的代码，发现跟之前《springBoot+mybatis数据库读写分离》这篇文章提到的方式二实现原理基本一样。
一些差异的地方有：
- 引入数据源分组的概念，这样方便支持一主多从的情况，更加充分利用从库资源
- 决定使用哪个数据源，是用DynamicDataSourceContextHolder类的peek方法来决定，重点就是peek！！！
DynamicDataSourceContextHolder
public final class DynamicDataSourceContextHolder {
/**
* 为什么要用链表存储(准确的是栈)
* <pre>
* 为了支持嵌套切换，如ABC三个service都是不同的数据源
* 其中A的某个业务要调B的方法，B的方法需要调用C的方法。一级一级调用切换，形成了链。
* 传统的只设置当前线程的方式不能满足此业务需求，必须模拟栈，后进先出。
* </pre>
*/
@SuppressWarnings("unchecked")
private static final ThreadLocal<Deque<String>> LOOKUP_KEY_HOLDER = new ThreadLocal() {
@Override
protected Object 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();
}
}
难得，这个类作者加了足够多的代码，一看就知道这个类为什么要这样设计，每个方法的作用。
这里比较重要的，就是ThreadLocal存的对象，从String，变成了ArrayDeque，这样，就可以在嵌套调用的时候，形成一个栈，要获取数据源的时候，就拿栈顶的元素，看看指定了哪个数据源。
但是我们思考一下，这样这的有意义么，没有问题么？？
作者想解决的问题是“A的某个业务要调B的方法，B的方法需要调用C的方法”而ABC各自需要访问的数据库不同。
假设A方法自己要往主库插入一条数据，得到一个id，B方法根据id从从库查询数据，C方法去从库查询其他的业务数据。
首先，我们考虑在使用事务的情况下，例如A方法添加了@Transactional注解：
1、spring的事务机制下，执行到A方法的时候，会获取一次connection。
2、执行到B方法时，当前线程变量里已经有动态数据源的connection，不会重新获取，使用已有的connection，则会继续从主库查询。
3、执行到C方法时，也是同样的道理，会从主库查询数据。
这样，作者所说的嵌套，根本不会起任何作用。
考虑一下非事务的情况：
1、执行到A方法，指定走主库，成功插入到了数据，得到id
2、执行到B方法，指定走从库，因为是deque，也没有事务，所以会重新拿connecion，而且能拿到slave从库。但是！！！用A得到的id去从库查数据，不一定查得到！！！主从同步延迟！！！
基于这些情况，个人觉得这个支持嵌套的功能，并不怎么合适用于读写分离的场景，而适合多数据源吧。但即使是多数据源，也要小时使用，必须使用@Transactional注解，不然就变成分布式事务。。。。。
个人认为，单纯的读写分离，更合适的做法是sharding-jdbc的方式：
只要走过一次主库，接下来的请求都走主库！！
这样，就不会遇到因为主从同步延迟而读不到数据的情况。
但是sharding-jdbc也有局限性，因为它实现的是，在同一个connection（即同一个事务中），只要走过一次主库，就会一直走主库。
而我们真正需要的是，在一个线程中，走过一次主库，就一直走主库。
dynamic-datasource-spring-boot-starter改进方向：
可以修改dynamic-datasource-spring-boot-starter的DynamicDataSourceContextHolder类的代码，再调用push方法的时候，判断Deque里是否有数据，有的话，最后一个数据是否是主库，是的话，强制改成入队主库。
sharding-jdbc改进方向：
增加一个自定义注解，和扫描这个注解的advisor，利用sharding-jdbc提供的hint功能，就可以让整个线程的sql语句都走指定的数据源。
DynamicDataSourceAnnotationAdvisor
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);
}
}
// 定义切入点，就是添加了@Ds注解的类和方法
private Pointcut buildPointcut() {
Pointcut cpc = new AnnotationMatchingPointcut(DS.class, true);
Pointcut mpc = AnnotationMatchingPointcut.forMethodAnnotation(DS.class);
return new ComposablePointcut(cpc).union(mpc);
}
}
这个类的作用很简单，就是定义切入点：添加了@Ds注解的类和方法，方法的优先级高于类。
DynamicDataSourceAnnotationInterceptor
public class DynamicDataSourceAnnotationInterceptor implements MethodInterceptor {
/**
* The identification of SPEL.
*/
private static final String DYNAMIC_PREFIX = "#";
private static final DynamicDataSourceClassResolver RESOLVER = new DynamicDataSourceClassResolver();
@Setter
private DsProcessor dsProcessor;
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
try {
// 拿到指定的数据源名称，添加到deque的头部
DynamicDataSourceContextHolder.push(determineDatasource(invocation));
return invocation.proceed();
} finally {
// 执行完方法，将指定的数据源名称移除，实现嵌套
DynamicDataSourceContextHolder.poll();
}
}
private String determineDatasource(MethodInvocation invocation) throws Throwable {
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();
// 如果指定的数据源以#开头，则用DsProcessor解析得到真正的数据源名称
// 如果不是#开头，则直接返回对应的值
return (!key.isEmpty() && key.startsWith(DYNAMIC_PREFIX)) ? dsProcessor
.determineDatasource(invocation, key) : key;
}
}
到这里，动态数据源的核心内容就已经分析完毕~~~
其他的一些类，大家有兴趣可以自己去看看源码。
类似资料：