如何写一个读写分离中间件-阿里云开发者社区

公司DBA一直埋怨atlas的难用，希望从客户端层出一个读写分离的方案。开源市场上在客户端做读写分离的开源软件基本上没有。业务方利用Spring自带的路由数据源能实现部分读写分离的功能，但是功能不够完善。部分参考sharingjdbc源码思想，利用部分业余时间，写了这个robustdb，总共只使用了十多个类，两千多行代码左右。

一、背景

随着业务量的增长，所有公司都不是直接选择分库分表设计方案的。很长一段时间内，会采用 库垂直拆分和分区表 来解决库表数据量比较大的问题，采用 读写分离 来解决访问压力比较大的问题。我们公司也是一样。目前绝大部分业务还是使用读写分离的方案。我相信很多公司和我们公司的架构一样，采用中间代理层做读写分离。结构如下：

代理层读写分离.png

第一层是VIP，通过VIP做中间映射层，避免了应用绑定数据库的真实IP，这样在数据库故障时，可以通过VIP飘移来将流量打到另一个库。但是VIP无法跨机房，为未来的异地多活设计埋下绕不过去的坎。
VIP下面一层是读写分离代理，我们公司使用的是360的Atlas。Atlas通过将SQL解析为DML（Data Modify Language）和DQL（Data Query Language），DML的请求全部发到主库，DQL根据配置比例分发到读库（读库包括主库和从库）。
使用Atlas不足的地方如下：
1）Altas不再维护更新，现存一些bug，bug网上很多描述；
2） atlas中没有具体应用请求IP与具体数据库IP之间的映射数据，所以无法准确查到访问DB的请求是来自哪个应用
3）Altas控制的粒度是sql语句，只能指定某条查询sql语句走主库，不能根据场景指定。
4） DB在自动关闭某个与atlas之间的连接时，atlas不会刷新，它仍有可能把这个失效的连接给下次请求的应用使用。
5）使用atlas，对后期增加其他功能模会比较麻烦。

基于Atlas以上问题，以及我们需要将数据库账号和连接配置集中管控。我们设计了下面这套方案：

客户端读写分离.png

通过在客户端做读写分离可以解决Atlas上面存在的不足。整个流程如下图所示：

image.png

二、robustdb原理

1、读写分离设计核心点--路由

支持每条sql按照DML、DQL类型的默认路由
需求描述：目前公司采用读写分离的方案来增强数据库的性能，所有的DML（insert、updata、delete）操作在主库，通过mysql的binlog同步，将数据同步到多个读库。所有的DQL（select）操作主库或从库，从而增强数据的读能力。
支持方法级别的指定路由
需求：在service中指定方法中所有db操作方法操作同一个数据库（主要是主库），保证方法中的db读写都操作主库，避免数据同步延迟导致读从库数据异常。从而保证整个方法的事务属性。

解决思路：
我们将获取真实数据库（主库还是哪个从库）放到需要建立连接时的地方，为此我们创建了BackendConnection（传统是先连接数据库，然后再创建连接，数据库连接可以参考我另一篇文章“数据库连接详解”）。
在获取数据库连接时，通过对请求的sql进行解析和类型判别，识别为DML和DQL。如果时DML，则在线程的单sql线程本地变量上设置为“master”，DQL则设置为“slave”，为后续选择数据库提供选择参考。
如果要支持方法级别的事务（也就是整个方法的sql请求都发送到主库），需要借助拦截器，我们采用的是AspectJ方式的拦截器，关于AOP的原理，可以参考AOP原理。会拦截所有带有类型为dataSourceType的annotation的方法。在执行方法前，在线程的多sql线程本地变量上设置dataSourceType的name值（name值“master”代表走主库，name值为“slave”代表走从库）。线程的多sql线程本地变量为后续选择数据库提供选择参考。在方法执行完后，清理本地线程变量。
多sql线程本地变量的优先级高于单sql线程本地变量的优先级。

多sql线程本地变量	单sql线程本地变量	最终选库结果
master	master	master
master	slave	master
slave	master	slave
slave	salve	slave

image.png

注意点
本地线程变量要使用阿里包装的Ttl，防止用户在方法内部启动线程池，导致普通的线程本地变量丢失，从而导致选库异常。源于本地线程变量及扩展的原理可以参考我的文章“ThreadLocal及其扩展”
使用Ttl之后，需要在公司的JVM启动参数中增加“-javaagent:/{Path}/transmittable-thread-local-2.6.0-SNAPSHOT.jar”。原理就是在JVM启动时，加载transmittable-thread-local中的类替换逻辑，将以后的Runnable、Callable、ExecuteService等线程池相关类替换成增强后的TtlRunnable、TtlCallable、TtlExecuteService等。

下面展示一下时序图中类的核心代码，仅供参考：
DataSoueceAspect

@Aspect
@Component
public class DataSourceAspect{
    @Around("execution(* *(..)) && @annotation(dataSourceType)")
    public Object aroundMethod(ProceedingJoinPoint pjd, DataSourceType dataSourceType) throws Throwable {      DataSourceContextHolder.setMultiSqlDataSourceType(dataSourceType.name());
        Object result = pjd.proceed();
        DataSourceContextHolder.clearMultiSqlDataSourceType();
        return result;
    }
}

BackendConnection

public final class BackendConnection extends AbstractConnectionAdapter {

    private AbstractRoutingDataSource abstractRoutingDataSource;
    
    //用于缓存一条sql（可能对应多个statement）或者一次事务中的连接
    private final Map<String, Connection> connectionMap = new HashMap<String, Connection>();

    //构造函数
    public BackendConnection(AbstractRoutingDataSource abstractRoutingDataSource) {
        this.abstractRoutingDataSource = abstractRoutingDataSource;
    }

    @Override
    public PreparedStatement prepareStatement(String sql) throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql);
    }

    @Override
    public DatabaseMetaData getMetaData() throws SQLException {
        if(connectionMap == null || connectionMap.isEmpty()){
            return abstractRoutingDataSource.getResolvedDefaultDataSource().getConnection().getMetaData();
        }
        return fetchCachedConnection(connectionMap.keySet().iterator().next().toString()).get().getMetaData();
    }

    @Override
    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)
            throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql,resultSetType,resultSetConcurrency);
    }

    @Override
    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency,
            int resultSetHoldability) throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
    }

    @Override
    public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql, autoGeneratedKeys);
    }

    @Override
    public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql, columnIndexes);
    }

    @Override
    public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
        return getConnectionInternal(sql).prepareStatement(sql, columnNames);
    }

    @Override
    protected Collection<Connection> getConnections() {
        return connectionMap.values();
    }

    /**
     * 根据sql获取连接，对连接进行缓存
     * @param sql
     * @return
     * @throws SQLException
     */
    private Connection getConnectionInternal(final String sql) throws SQLException {
        //设置线程环境遍历
        if (ExecutionEventUtil.isDML(sql)) {
            DataSourceContextHolder.setSingleSqlDataSourceType(DataSourceType.MASTER);
        } else if (ExecutionEventUtil.isDQL(sql)) {
            DataSourceContextHolder.setSingleSqlDataSourceType(DataSourceType.SLAVE);
        }
        //根据上面设置的环境变量，选择相应的数据源
        Object dataSourceKey = abstractRoutingDataSource.determineCurrentLookupKey();
        String dataSourceName = dataSourceKey.toString();
        
        //看缓存中是否已经含有相应数据源的连接
        Optional<Connection> connectionOptional = fetchCachedConnection(dataSourceName);
        if (connectionOptional.isPresent()) {
            return connectionOptional.get();
        }
        //缓存中没有相应连接，建立相应连接，并放入缓存
        Connection connection = abstractRoutingDataSource.getTargetDataSource(dataSourceKey).getConnection();
        connection.setAutoCommit(super.getAutoCommit());
        connection.setTransactionIsolation(super.getTransactionIsolation());
        
        connectionMap.put(dataSourceKey.toString(), connection);
        
        return connection;
    }

    /**
     * 从缓存中取数据源
     * @param dataSourceName
     * @return
     */
    private Optional<Connection> fetchCachedConnection(final String dataSourceName) {
        if (connectionMap.containsKey(dataSourceName)) {
            return Optional.of(connectionMap.get(dataSourceName));
        }
        return Optional.absent();
    }
    
}

AbstractRoutingDataSource

/**
 * 
 * @Type AbstractRoutingDataSource
 * @Desc 数据源路由器（spring的AbstractRoutingDataSource将resolvedDataSources的注入放在bean初始化）
 * @Version V1.0
 */
public abstract class AbstractRoutingDataSource extends AbstractDataSource {
    
    private boolean lenientFallback = true;
    
    private Map<Object, Object> targetDataSources;

    private Object defaultTargetDataSource;
    
    private Map<Object, DataSource> resolvedDataSources = new HashMap<Object, DataSource>();
    
    private DataSource resolvedDefaultDataSource;
    
    private Logger logger = LoggerFactory.getLogger(AbstractRoutingDataSource.class);
                       

    public BackendConnection getConnection() throws SQLException {
        return new BackendConnection(this);
    }

    public BackendConnection getConnection(String username, String password)
            throws SQLException {
        return new BackendConnection(this);
        
    }
    
    public void afterPropertiesSet() {
        if (this.targetDataSources == null) {
            throw new IllegalArgumentException("Property 'targetDataSources' is required");
        }
        this.resolvedDataSources = new HashMap<Object, DataSource>(this.targetDataSources.size());
        for (Map.Entry entry : this.targetDataSources.entrySet()) {
            Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());
            DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());
            this.resolvedDataSources.put(lookupKey, dataSource);
        }
        if (this.defaultTargetDataSource != null) {
            this.resolvedDefaultDataSource = resolveSpecifiedDataSource(this.defaultTargetDataSource);
        }
    }

    public void putNewDataSource(Object key, DataSource dataSource){
        if(this.resolvedDataSources == null){
            this.resolvedDataSources = new HashMap<Object, DataSource>();
        }
        if(this.resolvedDataSources.containsKey(key)){
            this.resolvedDataSources.remove(key);
            logger.info("remove old key:" + key);
        }
        logger.info("add key:" + key + ", value=" + dataSource);
        this.resolvedDataSources.put(key, dataSource);
    }
    
    /**
     * 数据源选择逻辑
     */
    public DataSource determineTargetDataSource() {
        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
       
        Object lookupKey = determineCurrentLookupKey();
        DataSourceContextHolder.clearSingleSqlDataSourceType();
        
        int index = 0;
        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {
            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());
            index++;
        }
        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 + "]");
        }
        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());
        return dataSource;
    }
    
    public DataSource getTargetDataSource(Object lookupKey) {
        Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
       
        if(lookupKey == null){
            lookupKey = determineCurrentLookupKey();
        }
        DataSourceContextHolder.clearSingleSqlDataSourceType();
        
        int index = 0;
        for (Entry<Object, DataSource> element : resolvedDataSources.entrySet()) {
            logger.debug("myAbstractDS, index:" + index + ", key:" + element.getKey() + ", value:" + element.getValue().toString());
            index++;
        }
        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 + "]");
        }
        logger.debug("myAbstractDS, hit DS is " + dataSource.toString());
        return dataSource;
    }
    
    public abstract Object determineCurrentLookupKey();
    
    public abstract Object getCurrentSlaveKey();
    
    @Override
    public boolean isWrapperFor(Class<?> iface) throws SQLException {
        return (iface.isInstance(this) || determineTargetDataSource().isWrapperFor(iface));
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T unwrap(Class<T> iface) throws SQLException {
        if (iface.isInstance(this)){
            return (T) this;
        }
        return determineTargetDataSource().unwrap(iface);
    }
    
    protected Object resolveSpecifiedLookupKey(Object lookupKey) {
        return lookupKey;
    }
    
    protected DataSource resolveSpecifiedDataSource(Object dataSource) throws IllegalArgumentException {
        if (dataSource instanceof DataSource) {
            return (DataSource) dataSource;
        }
        else {
            throw new IllegalArgumentException(
                    "Illegal data source value - only [javax.sql.DataSource] and String supported: " + dataSource);
        }
    }
  //get set方法省略
}

DataSourceContextHolder

public class DataSourceContextHolder {
    
    private static final TransmittableThreadLocal<String> singleSqlContextHolder = new TransmittableThreadLocal<String>();
    
    private static final TransmittableThreadLocal<String> multiSqlContextHolder = new TransmittableThreadLocal<String>();
    
    /**
     * @Description: 设置单条sql数据源类型
     * @param dataSourceType  数据库类型
     * @return void
     * @throws
     */
    public static void setSingleSqlDataSourceType(String dataSourceType) {
        singleSqlContextHolder.set(dataSourceType);
    }
    
    /**
     * @Description: 获取单条sql数据源类型
     * @param 
     * @return String
     * @throws
     */
    public static String getSingleSqlDataSourceType() {
        return singleSqlContextHolder.get();
    }
    
    /**
     * @Description: 清除单条sql数据源类型
     * @param 
     * @return void
     * @throws
     */
    public static void clearSingleSqlDataSourceType() {
        singleSqlContextHolder.remove();
    }
    
    /**
     * @Description: 设置多条sql数据源类型
     * @param dataSourceType  数据库类型
     * @return void
     * @throws
     */
    public static void setMultiSqlDataSourceType(String dataSourceType) {
        multiSqlContextHolder.set(dataSourceType);
    }
    
    /**
     * @Description: 获取多条sql数据源类型
     * @param 
     * @return String
     * @throws
     */
    public static String getMultiSqlDataSourceType() {
        return multiSqlContextHolder.get();
    }
    
    /**
     * @Description: 清除多条sql数据源类型
     * @param 
     * @return void
     * @throws
     */
    public static void clearMultiSqlDataSourceType() {
        multiSqlContextHolder.remove();
    }

    /**
     * 判断当前线程是否为使用从库为数据源. 最外层service有slave的aop标签  或者 service没有aop标签且单条sql为DQL
     * 
     * @return
     */
    public static boolean isSlave() {
        return "slave".equals(multiSqlContextHolder.get()) || (multiSqlContextHolder.get()==null && "slave".equals(singleSqlContextHolder.get())) ;
    }  
    
}

DynamicDataSource

public class DynamicDataSource extends AbstractRoutingDataSource implements InitializingBean{  
    
    private static final Logger logger = LoggerFactory.getLogger(DynamicDataSource.class); 
    
    private Integer slaveCount = 0;  
    
    // 轮询计数,初始为-1,AtomicInteger是线程安全的  
    private AtomicInteger counter = new AtomicInteger(-1); 
    
    // 记录读库的key  
    private List<Object> slaveDataSources = new ArrayList<Object>(0); 
    
    // slave库的权重
    private  Map<Object,Integer>  slaveDataSourcesWeight;
    
    private Object currentSlaveKey;
    
    public DynamicDataSource() {
        super();
    }

    /**
     * 构造函数
     * @param defaultTargetDataSource
     * @param targetDataSources
     * @param slaveDataSourcesWeight
     */
    public DynamicDataSource(Object defaultTargetDataSource, Map<Object,Object> targetDataSources, Map<Object,Integer> slaveDataSourcesWeight) {
        this.setResolvedDataSources(new HashMap<Object, DataSource>(targetDataSources.size()));
        for (Map.Entry<Object, Object> entry : targetDataSources.entrySet()) {
            DataSource dataSource = resolveSpecifiedDataSource(entry.getValue());
            this.putNewDataSource(entry.getKey(), dataSource);
        }
        if (defaultTargetDataSource != null) {
            this.setResolvedDefaultDataSource(resolveSpecifiedDataSource(defaultTargetDataSource));
        }
        this.setSlaveDataSourcesWeight(slaveDataSourcesWeight);
        this.afterPropertiesSet();
    }

    @Override  
    public Object determineCurrentLookupKey() {  
        // 使用DataSourceContextHolder保证线程安全，并且得到当前线程中的数据源key  
        if (DataSourceContextHolder.isSlave()) {  
            currentSlaveKey = getSlaveKey();  
            return currentSlaveKey;  
        }  
        //TODO
        Object key = "master";  
        return key;  
    }  
    
    
    @Override  
    public void afterPropertiesSet() {  
        try {  
            super.afterPropertiesSet();
            Map<Object, DataSource> resolvedDataSources = this.getResolvedDataSources();  
            //清空从库节点，重新生成
            slaveDataSources.clear();
            slaveCount = 0;
            for (Map.Entry<Object, DataSource> entry : resolvedDataSources.entrySet()) { 
                if(slaveDataSourcesWeight.get(entry.getKey())==null){
                    continue;
                }
                for(int i=0; i<slaveDataSourcesWeight.get(entry.getKey());i++){
                    slaveDataSources.add(entry.getKey());  
                    slaveCount++;
                } 
            }  
        } catch (Exception e) {  
            logger.error("afterPropertiesSet error! ", e);  
        }  
    }  
  
    /** 
     * 轮询算法实现 
     *  
     * @return 
     */  
    public Object getSlaveKey() {  
        if(slaveCount <= 0 || slaveDataSources == null || slaveDataSources.size() <= 0){
            return null;
        }
        Integer index = counter.incrementAndGet() % slaveCount;  
        if (counter.get() > 9999) { // 以免超出Integer范围  
            counter.set(-1); // 还原  
        }  
        return slaveDataSources.get(index);  
    }

    public Map<Object, Integer> getSlaveDataSourcesWeight() {
        return slaveDataSourcesWeight;
    }

    public void setSlaveDataSourcesWeight(Map<Object, Integer> slaveDataSourcesWeight) {
        this.slaveDataSourcesWeight = slaveDataSourcesWeight;
    }

    public Object getCurrentSlaveKey() {
        return currentSlaveKey;
    }
}

2、读库流量分配策略设计

在我之前的文章“数据库智能云平台”提到，我们所有的数据库连接都是管控起来的，包括每个库的流量配置都是支持动态分配的。
支持读库按不同比例承接读请求。通过配置页面动态调整应用的数据库连接以及比例，支持随机或者顺序的方式将流量分配到相应的读库中去。
这里我们使用的配置管理下发中心是我们公司自己开发的gconfig，当然替换成开源的diamond或者applo也是可以的。
当接收到配管中心的调整指令，会动态更新应用数据源连接，然后更新beanFactory中的datasource。核心函数如下：

/**
     * 更新beanFactory
     * @param properties
     */
    public void refreshDataSource(String properties) {
        YamlDynamicDataSource dataSource;
        try {
            dataSource = new YamlDynamicDataSource(properties);
        } catch (IOException e) {
            throw new RuntimeException("convert datasource config failed!");
        }

        // 验证必须字段是否存在
        if (dataSource == null && dataSource.getResolvedDataSources() == null
                || dataSource.getResolvedDefaultDataSource() == null || dataSource.getSlaveDataSourcesWeight() == null) {
            throw new RuntimeException("datasource config error!");
        }
        ConcurrentHashMap<Object, DataSource> newDataSource = new ConcurrentHashMap<Object, DataSource>(
                dataSource.getResolvedDataSources());

        //更新数据源的bean
        DynamicDataSource dynamicDataSource = (DynamicDataSource) ((DefaultListableBeanFactory) beanFactory)
                .getBean(dataSourceName);
        dynamicDataSource.setResolvedDefaultDataSource(dataSource.getResolvedDefaultDataSource());
        dynamicDataSource.setResolvedDataSources(new HashMap<Object, DataSource>());//将数据源清空，重新添加
        for (Entry<Object, DataSource> element : newDataSource.entrySet()) {
            dynamicDataSource.putNewDataSource(element.getKey(), element.getValue());
        }
        dynamicDataSource.setSlaveDataSourcesWeight(dataSource.getSlaveDataSourcesWeight());
        dynamicDataSource.afterPropertiesSet();

    }

三、性能

我们经过性能测试，发现Robustdb的性能在一定层度上比Atlas性能更好。压测结果如下：

image.png

四、参考

https://tech.meituan.com/mtddl.html
https://tech.meituan.com/%E6%95%B0%E6%8D%AE%E5%BA%93%E9%AB%98%E5%8F%AF%E7%94%A8%E6%9E%B6%E6%9E%84%E7%9A%84%E6%BC%94%E8%BF%9B%E4%B8%8E%E8%AE%BE%E6%83%B3.html

如何写一个读写分离中间件

一、背景

二、robustdb原理

1、读写分离设计核心点--路由

2、读库流量分配策略设计

三、性能

四、参考

中间件

热门文章

最新文章

相关课程

相关电子书