文章目录
概述
随业务量增长，数据库读写分离是迟早要面临的问题。另外，公司在上规模后一般也会要求统一采用主从分布式数据库。
我习惯的处理方案是在应用层进行隔离：即将以写为主的业务放在一个应用上，以读为主的业务放在其他应用上。这应该算是最简单粗暴的解决方案了，却也能帮我应对90%需要读写分离的场景。不过总还有10%的特殊场景需要思考下怎样在应用内实现读写分离。
在应用内做读写分离大体上需要考虑三件事情：
- 多数据源实现
- 读写请求识别
- 读写请求分流
其中后两点是执行读写分离的关键。
接下来详细介绍下怎么在Springboot+MyBatis的应用中实现读写分离。这里会用到H2数据库和dbcp2数据库连接池。在测试中不会真的创建一个数据库集群，我们只需要能够验证写入和读取是访问的两个不同的数据库即可。
1. 多数据源实现
之前在《SpringBoot自定义数据源及多数据源配置》这篇文里我有介绍过怎样做多数据源实现。这次的做法也差不多。
下面是在配置文件中做的多数据源配置：
|
1 2 3 4 5 6 7 8 9 |
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中读取配置：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
@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的实例：
|
1 2 3 4 5 |
DataSource dsWrite = this.dsWriteCfg .initializeDataSourceBuilder() .type(BasicDataSource.class) .build(); |
这里只是想试试这种方案。按照老路子创建DataSource实例也是OK的，并且还会更简洁：
|
1 2 3 4 5 |
@Bean(name = "dsRead") @ConfigurationProperties(prefix = "datasource.read") public DataSource setDataSource() { return new BasicDataSource(); } |
至此多数据源配置已经完成。
2. 读写请求识别
也看过其他人的读写分离方案，其中读写请求识别这一层多是通过自定义注解+AOP来实现的。这种方案当然没问题，但是稍嫌有些繁琐，如果忘掉了添加注解就会导致意外。我更想要的是一种‘润物细无声’的实现。
之前做过一个为写入数据库的实例赋默认值的方案：《MyBatis写入时null问题统一处理方案》。这个方案的思路是通过自定义实现MyBatis拦截器来拦截写数据库请求并补上未赋值的数据。稍稍变通下，就可以改为拦截并识别查询语句：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
@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执行的存储。看下实现：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 |
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这个类。核心是下面这个方法：
|
1 2 3 4 5 6 7 8 9 10 11 12 |
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类并实现路由方案可以参考如下代码：
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
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！！
发表评论