sharding-jdbc讀寫分離原理詳細解析

更新時間：2023年12月15日 10:34:00 作者：那個天真的人

這篇文章主要介紹了sharding-jdbc讀寫分離原理詳細解析,很多時候,為了應付DB的高并發(fā)讀寫,我們會采用讀寫分離技術,讀寫分離指的是利用數(shù)據(jù)庫主從技術(把數(shù)據(jù)復制到多個節(jié)點中),分散讀多個庫以支持高并發(fā)的讀,需要的朋友可以參考下

前言

很多時候，為了應付DB的高并發(fā)讀寫，我們會采用讀寫分離技術。讀寫分離指的是利用數(shù)據(jù)庫主從技術（把數(shù)據(jù)復制到多個節(jié)點中），分散讀多個庫以支持高并發(fā)的讀，而寫只在master庫上。DB的主從技術只負責對數(shù)據(jù)進行復制和同步，而讀寫分離技術需要業(yè)務應用自身去實現(xiàn)。sharding-jdbc通過簡單的開發(fā)，可以方便的實現(xiàn)讀寫分離技術。本篇主要介紹其實現(xiàn)的原理。

sharding-jdbc讀寫分離特性說明

sharding-jdbc官方對其支持的讀寫分離技術進行了說明：

支持項 提供了一主多從的讀寫分離配置，可獨立使用，也可配合分庫分表使用。同個調用線程，執(zhí)行多條語句，其中一旦發(fā)現(xiàn)有非讀操作，后續(xù)所有讀操作均從主庫讀取。 Spring命名空間。基于Hint的強制主庫路由。

不支持范圍 主庫和從庫的數(shù)據(jù)同步。主庫和從庫的數(shù)據(jù)同步延遲導致的數(shù)據(jù)不一致。主庫雙寫或多寫。

簡單說明 sharding-jdbc實現(xiàn)讀寫分離技術的思路比較簡潔，不支持類似主庫雙寫或多寫這樣的特性，但目前來看，已經(jīng)可以滿足一般的業(yè)務需求了。

讀寫分離實現(xiàn)demo

庫和表的設計結構如下：

這里寫圖片描述

簡單的java代碼示例：

public final class MasterSlaveMain {
    public static void main(final String[] args) throws SQLException {
        DataSource dataSource = getShardingDataSource();
        printSimpleSelect(dataSource);
    }
    private static void printSimpleSelect(final DataSource dataSource) throws SQLException {
        String sql = "SELECT i.* FROM t_order o JOIN t_order_item i ON o.order_id=i.order_id WHERE o.user_id=? AND o.order_id=?";
        try (
                Connection conn = dataSource.getConnection();
                PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
            preparedStatement.setInt(1, 10);
            preparedStatement.setInt(2, 1001);
            try (ResultSet rs = preparedStatement.executeQuery()) {
                while (rs.next()) {
                    System.out.println(rs.getInt(1));
                    System.out.println(rs.getInt(2));
                    System.out.println(rs.getInt(3));
                }
            }
        }
    }
    private static ShardingDataSource getShardingDataSource() throws SQLException {
        DataSourceRule dataSourceRule = new DataSourceRule(createDataSourceMap());
        TableRule orderTableRule = TableRule.builder("t_order").actualTables(Arrays.asList("t_order_0", "t_order_1")).dataSourceRule(dataSourceRule).build();
        TableRule orderItemTableRule = TableRule.builder("t_order_item").actualTables(Arrays.asList("t_order_item_0", "t_order_item_1")).dataSourceRule(dataSourceRule).build();
        ShardingRule shardingRule = ShardingRule.builder().dataSourceRule(dataSourceRule).tableRules(Arrays.asList(orderTableRule, orderItemTableRule))
                .databaseShardingStrategy(new DatabaseShardingStrategy("user_id", new ModuloDatabaseShardingAlgorithm()))
                .tableShardingStrategy(new TableShardingStrategy("order_id", new ModuloTableShardingAlgorithm())).build();
        return new ShardingDataSource(shardingRule);
    }
    private static Map<String, DataSource> createDataSourceMap() throws SQLException {
        Map<String, DataSource> result = new HashMap<>(2, 1);
        Map<String, DataSource> slaveDataSourceMap1 = new HashMap<>(2, 1);
        slaveDataSourceMap1.put("ds_0_slave_0", createDataSource("ds_0_slave_0"));
        slaveDataSourceMap1.put("ds_0_slave_1", createDataSource("ds_0_slave_1"));
        result.put("ds_0", MasterSlaveDataSourceFactory.createDataSource("ds_0", "ds_0_master", createDataSource("ds_0_master"), slaveDataSourceMap1));
        Map<String, DataSource> slaveDataSourceMap2 = new HashMap<>(2, 1);
        slaveDataSourceMap2.put("ds_1_slave_0", createDataSource("ds_1_slave_0"));
        slaveDataSourceMap2.put("ds_1_slave_1", createDataSource("ds_1_slave_1"));
        result.put("ds_1", MasterSlaveDataSourceFactory.createDataSource("ds_1", "ds_1_master", createDataSource("ds_1_master"), slaveDataSourceMap2));
        return result;
    }
    private static DataSource createDataSource(final String dataSourceName) {
        BasicDataSource result = new BasicDataSource();
        result.setDriverClassName(com.mysql.jdbc.Driver.class.getName());
        result.setUrl(String.format("jdbc:mysql://localhost:3306/%s", dataSourceName));
        result.setUsername("root");
        result.setPassword("123456");
        return result;
    }
}

讀寫分離實現(xiàn)原理

一般我們是這樣來執(zhí)行sql語句的:

Connection conn = dataSource.getConnection();
PreparedStatement preparedStatement = conn.prepareStatement(sql);
preparedStatement.executeQuery();

這是利用原生jdbc操作數(shù)據(jù)庫查詢語句的一般流程，獲取一個連接，然后生成Statement，最后再執(zhí)行查詢。那么sharding-jdbc是在哪一塊進行擴展從而實現(xiàn)讀寫分離的呢？

想一下，想要實現(xiàn)讀寫分離，必然會涉及到多個底層的Connection，從而構造出不同連接下的Statement語句，而很多第三方軟件，如Spring，為了實現(xiàn)事務，調用dataSource.getConnection()之后，在一次請求過程中，可能就不會再次調用getConnection方法了，所以在dataSource.getConnection中做讀寫擴展是不可取的。為了更好的說明問題，看下面的例子：

Connection conn = getConnection();
PreparedStatement preparedStatement1 = conn.prepareStatement(sql1);
preparedStatement1.executeQuery();
Connection conn2 = getConnection();
PreparedStatement preparedStatement2 = conn2.prepareStatement(sql2);
preparedStatement2.executeUpdate();

一次請求過程中，為了實現(xiàn)事務，一般的做法是當線程第一次調用getConnection方法時，獲取一個底層連接，然后存儲到ThreadLocal變量中去，下次就直接在ThreadLocal中獲取了。為了實現(xiàn)一個事務中，針對一個數(shù)據(jù)源，既可能獲取到主庫連接，也可能獲取到從庫連接，還能夠切換，sharding-jdbc在PreparedStatement(實際上為ShardingPreparedStatement)的executeXX層進行了主從庫的連接處理。

下圖為sharding-jdbc執(zhí)行的部分流程：

這里寫圖片描述

sharding-jdbc使用ShardingPreparedStatement來替代PreparedStatement，在執(zhí)行ShardingPreparedStatement的executeXX方法時，通過路由計算，得到PreparedStatementUnit單元列表，然后執(zhí)行后合并結果返回，而PreparedStatementUnit只不過封裝了原生的PreparedStatement。讀寫分離最關鍵的地方在上圖標綠色的地方，也就是生成PreparedStatement的地方。

在使用SQLEcecutionUnit轉換為PreparedStatement的時候，有一個重要的步驟就是必須先獲取Connection，源碼如下：

public Connection getConnection(final String dataSourceName, final SQLType sqlType) throws SQLException {
        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;
        if (dataSource instanceof MasterSlaveDataSource) {
            NamedDataSource namedDataSource = ((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
            realDataSourceName = namedDataSource.getName();
            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;
    }

如果發(fā)現(xiàn)數(shù)據(jù)源對象為MasterSlaveDataSource類型，則會使用如下方式獲取真正的數(shù)據(jù)源：

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 static boolean isMasterRoute(final SQLType sqlType) {
    return SQLType.DQL != sqlType || DML_FLAG.get() || HintManagerHolder.isMasterRouteOnly();
}

有三種情況會認為一定要走主庫：

1. 不是查詢類型的語句，比如更新字段

2. DML_FLAG變量為true的時候

3. 強制Hint方式走主庫

當執(zhí)行了更新語句的時候，isMasterRoute()==true，這時候，Connection為主庫的連接，并且引擎會強制設置DML_FLAG的值為true，這樣一個請求后續(xù)的所有讀操作都會走主庫。有些時候，我們想強制走主庫，這時候在請求最開始執(zhí)行Hint操作即可，如下所示：

HintManager hintManager = HintManager.getInstance();
hintManager.setMasterRouteOnly();

在獲取數(shù)據(jù)源的時候，如果走的是從庫，會使用從庫負載均衡算法類進行處理，該類的實現(xiàn)比較簡單，如下所示：

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) {
        AtomicInteger count = COUNT_MAP.containsKey(name) ? COUNT_MAP.get(name) : new AtomicInteger(0);
        COUNT_MAP.putIfAbsent(name, count);
        count.compareAndSet(slaveDataSourceNames.size(), 0);
        return slaveDataSourceNames.get(count.getAndIncrement() % slaveDataSourceNames.size());
    }
}

其實就是一個簡單的輪循機制進行從庫的負載均衡。