Spring动态数据源+Mybatis拦截器实现数据库读写分离

Posted 2021-05-02 夜猫故事

在项目中遇到了需要做读写分离的场景。
对于老项目来说，尽量减少代码入侵，在底层实现读写分离是坠吼的。

用到的技术主要有两点：

• spring动态数据源

• mybatis拦截器

spring动态数据源

对于多数据源的情况，spring提供了动态数据源

org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource

动态数据源可以通过配置key值，来获取对应的不同的数据源。

但是要注意一点：动态数据源不是真正的数据源！

AbstractRoutingDataSource 正如其名，只是提供了数据源路由的功能，具体的数据源还需要进行单独的配置。所以在我们的实现中，还需要对数据源的配置和生成进行实现。

数据源的配置还是十分简单的，在实现类DynamicDataSource中，声明了三组数据源集合：

 //直接给定数据源
    private List<DataSource> roDataSources;    private List<DataSource> woDataSources;    private List<DataSource> rwDataSources;

使用时通过spring注入配置好的数据源，然后遍历三个集合，根据配置给指定不同的key。
为了统一进行key的管理，将数据源key的生成和指派都放在了一个单例的OPCountMapper类中进行管理，此类中根据数据源所在集合，分别给定只读，读写和只写三种key以及编号，在进行操作时根据操作的类型，依次调用每一种key中的每个数据源。也就是自带简单的负载均衡功能。

import static com.kingsoft.multidb.MultiDbConstants.*;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;/** * 数据源key管理 * 每个数据源对应一个单独的的key例如： * ro_0,rw_1,wo_2 * 之类。 * 本映射类通过操作类型选定一个可用的数据库进行操作。 * 当对应类型没有可用数据源时，使用读写数据源。 * Created by SHIZHIDA on 2017/7/4. */public class OPCountMapper {    private Map<String,Integer> countMapper = new ConcurrentHashMap<>();    private Map<String,Integer> lastRouter = new ConcurrentHashMap<>();    public OPCountMapper(){
        countMapper.put(RO,0);
        countMapper.put(RW,0);
        countMapper.put(WO,0);
        lastRouter.put(RO,0);
        lastRouter.put(RW,0);
        lastRouter.put(WO,0);
    }    public String getCurrentRouter(String key){        int total = countMapper.get(key);        if(total==0){            if(!key.equals(RW))                return getCurrentRouter(RW);            else{                return null;
            }
        }        int last = lastRouter.get(key);        return key+"_"+(last+1)%total;
    }    public String appendRo() {        return appendKey(RO);
    }    public String appendWo() {        return appendKey(WO);
    }    public String appendRw() {        return appendKey(RW);
    }    private String appendKey(String key){        int total = countMapper.get(key);        String sk = key+"_"+total++;
        countMapper.put(key,total);        return sk;
    }
}

最后则是在使用中指定当前数据源，这里利用到java的ThreadLocal类。此类为每一个线程维护一个单独的成员变量。在使用时，可以根据当前的操作，指定此线程中需要使用的数据源类型：

/** * 数据库选择 * Created by SHIZHIDA on 2017/7/4. */public final class DataSourceSelector {    private static ThreadLocal<String> currentKey = new ThreadLocal<>();    public static String getCurrentKey(){        String key = currentKey.get();        if(StringUtils.isNotEmpty(key))            return key;        else return RW;
    }    public static void setRO(){
        setCurrenKey(RO);
    }    public static void setRW(){
        setCurrenKey(RW);
    }    public static void setWO(){
        setCurrenKey(WO);
    }    public static void setCurrenKey(String key){        if(Arrays.asList(RO,WO,RW).indexOf(key)>=0){
            currentKey.set(key);
        }else{
            currentKey.set(RW);
            warn("undefined key:"+key);
        }
    }
    
}

Mybatis拦截器

上面讲述了数据源的配置和选择，那么进行选择的功能就交给Mybatis的拦截器来实现了。

首先，Mybatis所有的SQL读写操作，都是通过 org.apache.ibatis.executor.Executor 类来进行操作的。追踪代码可发现，这个类中读写只有三个接口，而且功能一目了然：

int update(MappedStatement ms, Object parameter) throws SQLException;

  <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler, CacheKey cacheKey, BoundSql boundSql) throws SQLException;

  <E> List<E> query(MappedStatement ms, Object parameter, RowBounds rowBounds, ResultHandler resultHandler) throws SQLException;

也就是说只要监控了这三个接口，就可以对所有的读写操作指派相应的数据源。

代码也十分简单：

import com.kingsoft.multidb.datasource.DataSourceSelector;import org.apache.ibatis.cache.CacheKey;import org.apache.ibatis.executor.Executor;import org.apache.ibatis.mapping.BoundSql;import org.apache.ibatis.mapping.MappedStatement;import org.apache.ibatis.plugin.*;import org.apache.ibatis.session.ResultHandler;import org.apache.ibatis.session.RowBounds;import java.util.Properties;/** * 拦截器，对update使用写库，对query使用读库 * Created by SHIZHIDA on 2017/7/4. */@Intercepts({        @Signature(
                type= Executor.class,
                method = "update",
                args = {MappedStatement.class,Object.class}),        @Signature(
                type= Executor.class,
                method = "query",
                args = {MappedStatement.class,Object.class, RowBounds.class, ResultHandler.class,CacheKey.class,BoundSql.class}),        @Signature(
                type= Executor.class,
                method = "query",
                args = {MappedStatement.class,Object.class,RowBounds.class, ResultHandler.class}),
})
public class DbSelectorInterceptor implements Interceptor {    @Override
    public Object intercept(Invocation invocation) throws Throwable {        String name = invocation.getMethod().getName();        if(name.equals("update"))
            DataSourceSelector.setWO();        if(name.equals("query"))
            DataSourceSelector.setRO();        return invocation.proceed();
    }    @Override
    public Object plugin(Object target) {        if(target instanceof Executor)            return Plugin.wrap(target,this);        else return target;
    }    @Override
    public void setProperties(Properties properties) {

    }

}

总结

至此一套简单的数据库读写分离功能就已经实现了，只要在spring中配置了数据源，并且为mybatis的SqlSessionFactory进行如下配置：

<bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
    <property name="dataSource" ref="dynamicDataSource"/>
    <property name="configLocation" value="classpath:mybatis-config.xml"/>
    <property name="plugins" ref="dbSelectorInterceptor"/>
</bean>

就可以在对代码0侵入的情况下实现读写分离，附赠多数据库负载均衡的功能。