概述

随业务量增长，数据库读写分离是迟早要面临的问题。另外，公司在上规模后一般也会要求统一采用主从分布式数据库。

我习惯的处理方案是在应用层进行隔离：即将以写为主的业务放在一个应用上，以读为主的业务放在其他应用上。这应该算是最简单粗暴的解决方案了，却也能帮我应对90%需要读写分离的场景。不过总还有10%的特殊场景需要思考下怎样在应用内实现读写分离。

在应用内做读写分离大体上需要考虑三件事情：

1. 多数据源实现
2. 读写请求识别
3. 读写请求分流

其中后两点是执行读写分离的关键。

接下来详细介绍下怎么在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决定了为之后的请求提供哪个数据源。根据代码可以看出来，我们至少需要做两件事：

1. 为resolvedDataSources赋值，即设置相关数据源
2. 实现分流方法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！！

参考文档

1. mybatis读写分离