16. sjdbc之读写分离

Posted 2021-05-02 阿飞的博客

读写分离支持范围

• 提供了一主多从的读写分离配置，可独立使用，也可配合分库分表使用。

• 同一线程且同一数据库连接内，如有写入操作，以后的读操作均从主库读取，用于保证数据一致性。

• Spring命名空间。

• 基于Hint的强制主库路由。

读写分离不支持范围

• 主库和从库的数据同步

• 主库和从库的数据同步延迟导致的数据不一致。

• 主库双写或多写。

---

读写分离支持项和不支持范围摘自sharding-jdbc使用指南☞读写分离：http://shardingjdbc.io/docs_1.x/02-guide/master-slave/

简单总结

sharding-jdbc的读写分离不甘于数据库层面的东西。比如主从同步，依赖mysql本身的同步机制，而不是由sharding-jdbc去支持。而主从同步延迟这种问题，sharding-jdbc也没必要去解决。

源码分析

MasterSlave or Sharding

选择MasterSlaveDataSource还是普通的ShardingDataSource逻辑非常简单，就看根据sharding配置能否得到slave，核心源码如下：

// 数据源名称与数据库连接关系缓存，例如：{dbtbl_0_master:Connection实例; dbtbl_1_master:Connection实例; dbtbl_0_slave_0:Connection实例; dbtbl_0_slave_1:Connection实例; dbtbl_1_slave_0:Connection实例; dbtbl_1_slave_1:Connection实例}
private final Map<String, Connection> cachedConnections = new HashMap<>();

/**
 * 根据数据源名称得到数据库连接
 */
public Connection getConnection(final String dataSourceName, final SQLType sqlType) throws SQLException {
    // 首先尝试从local cache（map类型）中获取，如果已经本地缓存，那么直接从本地缓存中获取
    if (getCachedConnections().containsKey(dataSourceName)) {
        return getCachedConnections().get(dataSourceName);
    }
    DataSource dataSource = shardingContext.getShardingRule().getDataSourceRule().getDataSource(dataSourceName);
    Preconditions.checkState(null != dataSource, "Missing the rule of %s in DataSourceRule", dataSourceName);
    String realDataSourceName;
    // 如果是主从数据库的话（例如xml中配置<rdb:master-slave-data-source id="dbtbl_0" ...>，那么dbtbl_0就是主从数据源）
    if (dataSource instanceof MasterSlaveDataSource) {
        // 见后面的"主从数据源中根据负载均衡策略获取数据源"的分析
        NamedDataSource namedDataSource = ((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
        realDataSourceName = namedDataSource.getName();
        // 如果主从数据库元选出的数据源名称（例如：dbtbl_1_slave_0）与数据库连接已经被缓存，那么从缓存中取出数据库连接
        if (getCachedConnections().containsKey(realDataSourceName)) {
            return getCachedConnections().get(realDataSourceName);
        }
        dataSource = namedDataSource.getDataSource();
    } else {
        realDataSourceName = dataSourceName;
    }
    Connection result = dataSource.getConnection();
    // 把数据源名称与数据库连接实例缓存起来
    getCachedConnections().put(realDataSourceName, result);
    replayMethodsInvocation(result);
    return result;
}

强制主&负载均衡选择

主从数据源中根据负载均衡策略获取数据源核心源码--MasterSlaveDataSource.java：

// 主数据源, 例如dbtbl_0_master对应的数据源
@Getter
private final DataSource masterDataSource;

// 主数据源下所有的从数据源，例如{dbtbl_0_slave_0:DataSource实例; dbtbl_0_slave_1:DataSource实例}
@Getter
private final Map<String, DataSource> slaveDataSources;

public NamedDataSource getDataSource(final SQLType sqlType) {
    if (isMasterRoute(sqlType)) {
        DML_FLAG.set(true);
        // 如果符合主路由规则，那么直接返回主路由（不需要根据负载均衡策略选择数据源）
        return new NamedDataSource(masterDataSourceName, masterDataSource);
    }
    // 负载均衡策略选择数据源名称[后面会分析]
    String selectedSourceName = masterSlaveLoadBalanceStrategy.getDataSource(name, masterDataSourceName, new ArrayList<>(slaveDataSources.keySet()));
    DataSource selectedSource = selectedSourceName.equals(masterDataSourceName) ? masterDataSource : slaveDataSources.get(selectedSourceName);
    Preconditions.checkNotNull(selectedSource, "");
    return new NamedDataSource(selectedSourceName, selectedSource);
}

// 主路由逻辑
private boolean isMasterRoute(final SQLType sqlType) {
    return SQLType.DQL != sqlType || DML_FLAG.get() || HintManagerHolder.isMasterRouteOnly();
}

主路由逻辑如下：

1. 非查询SQL（SQLType.DQL != sqlType）

2. 当前数据源在当前线程访问过主库（数据源访问过主库就会通过ThreadLocal将DML_FLAG置为true，从而路由主库）（DML_FLAG.get()）

3. HintManagerHolder方式设置了主路由规则（HintManagerHolder.isMasterRouteOnly()）

---

当前线程访问过主库后，后面的操作全部切主，是为了防止主从同步数据延迟导致写操作后，读不到最新的数据？我想应该是这样的^^

主从负载均衡分析

从对MasterSlaveDataSource.java的分析可知，如果不符合强制主路由规则，那么会根据负载均衡策略选多个slave中选取一个slave，MasterSlaveLoadBalanceAlgorithm接口有两个实现类：

• RoundRobinMasterSlaveLoadBalanceAlgorithm即轮训策略；

• RandomMasterSlaveLoadBalanceAlgorithm随机策略；

笔者有话要说：这里没有出镜率比较高的一致性hash策略，这倒是个意外，哈！

---

轮询策略

轮询方式的实现类为RoundRobinMasterSlaveLoadBalanceStrategy，核心源码如下：

public final class RoundRobinMasterSlaveLoadBalanceStrategy implements MasterSlaveLoadBalanceStrategy {

    private static final ConcurrentHashMap<String, AtomicInteger> COUNT_MAP = new ConcurrentHashMap<>();

    @Override
    public String getDataSource(final String name, final String masterDataSourceName, final List<String> slaveDataSourceNames) {
        // 每个集群体系都有自己的计数器，例如dbtbl_0集群，dbtbl_1集群；如果COUNT_MAP中还没有这个集群体系，需要先初始化；
        AtomicInteger count = COUNT_MAP.containsKey(name) ? COUNT_MAP.get(name) : new AtomicInteger(0);
        COUNT_MAP.putIfAbsent(name, count);
        // 如果轮询计数器（AtomicInteger count）长到slave.size()，那么归零（防止计数器不断增长下去）
        count.compareAndSet(slaveDataSourceNames.size(), 0);
        // 计数器递增，根据计算器的值对slave集群数值取模就是从slave集合中选中的目标slave的下标
        return slaveDataSourceNames.get(count.getAndIncrement() % slaveDataSourceNames.size());
    }
}

---

注意轮询选择最后一行代码：

return slaveDataSourceNames.get(count.getAndIncrement() % slaveDataSourceNames.size())

count.getAndIncrement()然后取模，而count的初始值为0。这就有一个非常小的隐患，当轮询策略选择了Integer.MAX_VALUE次后，再count.getAndIncrement()就会变成负数，从而导致执行List集合的get()方法出错。验证这个潜在问题的代码如下：

public class SlaveDataSourceSelectDemo {

    public static void main(String[] args) {

        List<String> dataSources = new ArrayList<>();
        dataSources.add("datasource-1");
        dataSources.add("datasource-2");
        dataSources.add("datasource-3");

        AtomicInteger count = new AtomicInteger(Integer.MAX_VALUE-3);
        for (int i = 0; i < 5; i++) {
            System.out.println("selected: "+dataSources.get(count.getAndIncrement() % dataSources.size()));
        }
    }
}

模拟主从总计3个节点，当轮询Integer.MAX_VALUE-3后，再尝试轮询5次，在第4次就会报错！

笔者在github上找到了sharding-sphere最新的代码，这个问题依然存在，明天找张亮大神确认一下，最新源码路径为：
https://github.com/sharding-sphere/sharding-sphere/blob/dev/sharding-core/src/main/java/io/shardingsphere/core/api/algorithm/masterslave/RoundRobinMasterSlaveLoadBalanceAlgorithm.java

随机策略

随机方式的实现类为RandomMasterSlaveLoadBalanceAlgorithm，核心源码如下：

public final class RandomMasterSlaveLoadBalanceAlgorithm implements MasterSlaveLoadBalanceAlgorithm {    
    @Override
    public String getDataSource(final String name, final String masterDataSourceName, final List<String> slaveDataSourceNames) {
        // 从实例范围取一个随机数（例如总计有3个slave实例，那么new Random().nextInt(3)），就是从slave集合中选中的目标slave的下标
        return slaveDataSourceNames.get(new Random().nextInt(slaveDataSourceNames.size()));
    }
}

默认策略

public enum MasterSlaveLoadBalanceAlgorithmType {

    ROUND_ROBIN(new RoundRobinMasterSlaveLoadBalanceAlgorithm()),
    RANDOM(new RandomMasterSlaveLoadBalanceAlgorithm());

    private final MasterSlaveLoadBalanceAlgorithm algorithm;

    // 默认策略就是轮询策略
    public static MasterSlaveLoadBalanceAlgorithmType getDefaultAlgorithmType() {
        return ROUND_ROBIN;
    }
}

---