随业务量增长，数据库读写分离是迟早要面临的问题。另外，公司在上规模后一般也会要求统一采用主从分布式数据库。
我习惯的处理方案是在应用层进行隔离：即将以写为主的业务放在一个应用上，以读为主的业务放在其他应用上。这应该算是最简单粗暴的解决方案了，却也能帮我应对90%需要读写分离的场景。不过总还有10%的特殊场景需要思考下怎样在应用内实现读写分离。
在应用内做读写分离大体上需要考虑三件事情：
- 多数据源实现
- 读写请求识别
- 读写请求分流
其中后两点是执行读写分离的关键。
接下来详细介绍下怎么在Springboot+MyBatis的应用中实现读写分离。这里会用到H2数据库和dbcp2数据库连接池。在测试中不会真的创建一个数据库集群，我们只需要能够验证写入和读取是访问的两个不同的数据库即可。
1. 多数据源实现
之前在《SpringBoot自定义数据源及多数据源配置》这篇文里我有介绍过怎样做多数据源实现。这次的做法也差不多。
下面是在配置文件中做的多数据源配置：
datasource:
write:
driver: org.h2.Driver
url: jdbc:h2:mem:worker-write
validation-query: select 1
read:
driver: org.h2.Driver
url: jdbc:h2:mem:worker-read
validation-query: select 1
这里配置了两个H2的内存数据库，我们权且当它们是一个集群吧。
然后是在一个配置类
DsConfig中读取配置：
@Configuration
public class DsConfig {
@Bean(name = "readCfg")
@ConfigurationProperties("datasource.read")
public DataSourceProperties readConfig() {
return new DataSourceProperties();
@Primary
@Bean(name = "writeCfg")
@ConfigurationProperties("datasource.write")
public DataSourceProperties writeConfig() {
return new DataSourceProperties();
和之前那篇文章《SpringBoot自定义数据源及多数据源配置》略有不同，这次是用
DataSourceProperties来表示读取的配置信息。
注意不要忽略了
@Primary注解，不然会报错。
可以这样使用
DataSourceProperties创建DataSource的实例：
DataSource dsWrite =
this.dsWriteCfg
.initializeDataSourceBuilder()
.type(BasicDataSource.class)
.build();
这里只是想试试这种方案。按照老路子创建
DataSource实例也是OK的，并且还会更简洁：
@Bean(name = "dsRead")
@ConfigurationProperties(prefix = "datasource.read")
public DataSource setDataSource() {
return new BasicDataSource();
至此多数据源配置已经完成。
2. 读写请求识别
也看过其他人的读写分离方案，其中读写请求识别这一层多是通过自定义注解+AOP来实现的。这种方案当然没问题，但是稍嫌有些繁琐，如果忘掉了添加注解就会导致意外。我更想要的是一种‘润物细无声’的实现。
之前做过一个为写入数据库的实例赋默认值的方案：《MyBatis写入时null问题统一处理方案》。这个方案的思路是通过自定义实现MyBatis拦截器来拦截写数据库请求并补上未赋值的数据。稍稍变通下，就可以改为拦截并识别查询语句：
@Component
@Intercepts({
@Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class}),
@Signature(type = Executor.class, method = "query", args = {MappedStatement.class, Object.class, RowBounds.class, ResultHandler.class, CacheKey.class, BoundSql.class})
public class MybatisReadInterceptor implements Interceptor {
@Override
public Object intercept(Invocation invocation) throws Throwable {
DsContextHolder.set(READ);
return invocation.proceed();
@Override
public Object plugin(Object target) {
return Plugin.wrap(target, this);
@Override
public void setProperties(Properties properties) {
查询请求主要是由
Executor类的
query方法实现的，所以只要针对其进行拦截并执行判断即可。
因为判断读写请求和执行读写分流是在两个环节执行，我们需要找个地方将判断结果存储起来，并且保证线程安全，很自然可以想到使用
ThreadLocal执行存储。
DsContextHolder就是基于
ThreadLocal执行的存储。看下实现：
public final class DsContextHolder {
private static ThreadLocal<DsType> context = new ThreadLocal<>();
public static void set(DsType type) {
if (null != type) {
context.set(type);
public static DsType getDbType() {
DsType type = context.get();
return (null == type ? WRITE : type);
public static void clear() {
context.remove();
private DsContextHolder() {
throw new UnsupportedOperationException("Private constructor, cannot be accessed!");
接下来就可以使用
DsContextHolder中存储的信息来执行分流了。
3. 读写请求分流
读写请求分流这一层主要依赖了
AbstractRoutingDataSource这个类。核心是下面这个方法：
protected DataSource determineTargetDataSource() {
Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
Object lookupKey = determineCurrentLookupKey();
DataSource dataSource = this.resolvedDataSources.get(lookupKey);
if (dataSource == null && (this.lenientFallback || lookupKey == null)) {
dataSource = this.resolvedDefaultDataSource;
if (dataSource == null) {
throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");
return dataSource;
看名字也能知道，
determineTargetDataSource决定了为之后的请求提供哪个数据源。根据代码可以看出来，我们至少需要做两件事：
- 为
resolvedDataSources赋值，即设置相关数据源
- 实现分流方法
determineCurrentLookupKey()
具体如何继承
AbstractRoutingDataSource类并实现路由方案可以参考如下代码：
public class ReadWriteDsRouter extends AbstractRoutingDataSource {
@Autowired
@Qualifier("readCfg")
private DataSourceProperties dsReadCfg;
@Autowired
@Qualifier("writeCfg")
private DataSourceProperties dsWriteCfg;
@Override
protected Object determineCurrentLookupKey() {
DsType type = DsContextHolder.getDbType();
DsContextHolder.clear();
return type;
@Override
public void afterPropertiesSet() {
DataSource dsWrite = this.dsWriteCfg.initializeDataSourceBuilder().type(BasicDataSource.class).build();
DataSource dsRead = this.dsReadCfg.initializeDataSourceBuilder().type(BasicDataSource.class).build();
Map<Object, Object> dataSources = new HashMap<>(2);
dataSources.put(READ, dsRead);
dataSources.put(WRITE, dsWrite);
this.setTargetDataSources(dataSources);
this.setDefaultTargetDataSource(dsWrite);
super.afterPropertiesSet();
在
afterPropertiesSet()方法中完成了自定义数据源的创建和设置，并且还将
dsWrite设置为了默认数据源。
方法
determineCurrentLookupKey()基于
DsContextHolder中存储的内容提供了分流的关键字。
大体上就是这样了。具体实现代码已经上传到了GitHub ： zhyea/database-wr
End！！
- mybatis读写分离
