資料庫中介軟體 Sharding-JDBC 原始碼分析 —— JDBC實現與讀寫分離

阿新 • • 發佈：2022-05-02

本文主要基於 Sharding-JDBC 1.5.0 正式版

1. 概述
2. unspported 包
3. adapter 包
3.1 WrapperAdapter
3.2 AbstractDataSourceAdapter
3.3 AbstractConnectionAdapter
3.4 AbstractStatementAdapter
3.5 AbstractPreparedStatementAdapter
3.6 AbstractResultSetAdapter
4. 插入流程
5. 查詢流程
6. 讀寫分離

1. 概述

本文主要分享 JDBC 與 讀寫分離 的實現。為什麼會把這兩個東西放在一起講呢？客戶端直連資料庫的讀寫分離主要通過獲取讀庫和寫庫的不同連線來實現，和 JDBC Connection 剛好放在一塊。

OK，我們先來看一段 Sharding-JDBC 官方對自己的定義和定位

Sharding-JDBC定位為輕量級java框架，使用客戶端直連資料庫，以jar包形式提供服務，未使用中間層，無需額外部署，無其他依賴，DBA也無需改變原有的運維方式，可理解為增強版的JDBC驅動，舊程式碼遷移成本幾乎為零。

可以看出，Sharding-JDBC 通過實現 JDBC規範，對上層提供透明化資料庫分庫分表的訪問。? 黑科技？實際我們使用的資料庫連線池也是通過這種方式實現對上層無感知的提供連線池。甚至還可以通過這種方式實現對 Lucene、MongoDB 等等的訪問。

扯遠了，下面來看看 Sharding-JDBC jdbc

包的結構：

unsupported：宣告不支援的資料操作方法
adapter：適配類，實現和分庫分表無關的方法
core：核心類，實現和分庫分表相關的方法

根據 core 包，可以看出分到四種我們超級熟悉的物件

Datasource

Connection

Statement

ResultSet

實現層級如下：JDBC 介面 <=(繼承)== unsupported抽象類 <=(繼承)== unsupported抽象類 <=(繼承)== core類。

本文內容順序

unspported 包
adapter 包
插入流程，分析的類：
- ShardingDataSource
- ShardingConnection
- ShardingPreparedStatement（ShardingStatement 類似，不重複分析）
- GeneratedKeysResultSet、GeneratedKeysResultSetMetaData
查詢流程，分析的類：
- ShardingPreparedStatement
- ShardingResultSet
讀寫分離，分析的類：
- MasterSlaveDataSource

2. unspported 包

unspported 包內的抽象類，宣告不支援操作的資料物件，所有方法都是 thrownewSQLFeatureNotSupportedException() 方式。

public abstract class AbstractUnsupportedGeneratedKeysResultSet extends AbstractUnsupportedOperationResultSet {

    @Override
    public boolean getBoolean(final int columnIndex) throws SQLException {
        throw new SQLFeatureNotSupportedException("getBoolean");
    }

    // .... 省略其它類似方法
}

public abstract class AbstractUnsupportedOperationConnection extends WrapperAdapter implements Connection {

    @Override
    public final CallableStatement prepareCall(final String sql) throws SQLException {
        throw new SQLFeatureNotSupportedException("prepareCall");
    }

   // .... 省略其它類似方法
}

3. adapter 包

adapter 包內的抽象類，實現和分庫分表無關的方法。

考慮到第4、5兩小節更容易理解，本小節貼的程式碼會相對多

3.1 WrapperAdapter

WrapperAdapter，JDBC Wrapper 適配類。

對 Wrapper 介面實現如下兩個方法：

@Override
public final <T> T unwrap(final Class<T> iface) throws SQLException {
   if (isWrapperFor(iface)) {
       return (T) this;
   }
   throw new SQLException(String.format("[%s] cannot be unwrapped as [%s]", getClass().getName(), iface.getName()));
}

@Override
public final boolean isWrapperFor(final Class<?> iface) throws SQLException {
   return iface.isInstance(this);
}

提供子類 #recordMethodInvocation() 記錄方法呼叫， #replayMethodsInvocation() 回放記錄的方法呼叫：

/**
* 記錄的方法陣列
*/
private final Collection<JdbcMethodInvocation> jdbcMethodInvocations = new ArrayList<>();

/**
* 記錄方法呼叫.
* 
* @param targetClass 目標類
* @param methodName 方法名稱
* @param argumentTypes 引數型別
* @param arguments 引數
*/
public final void recordMethodInvocation(final Class<?> targetClass, final String methodName, final Class<?>[] argumentTypes, final Object[] arguments) {
   try {
       jdbcMethodInvocations.add(new JdbcMethodInvocation(targetClass.getMethod(methodName, argumentTypes), arguments));
   } catch (final NoSuchMethodException ex) {
       throw new ShardingJdbcException(ex);
   }
}

/**
* 回放記錄的方法呼叫.
* 
* @param target 目標物件
*/
public final void replayMethodsInvocation(final Object target) {
   for (JdbcMethodInvocation each : jdbcMethodInvocations) {
       each.invoke(target);
   }
}

這兩個方法有什麼用途呢？例如下文會提到的 AbstractConnectionAdapter 的 #setAutoCommit()，當它無資料庫連線時，先記錄；等獲得到資料連線後，再回放：

// AbstractConnectionAdapter.java
@Override
public final void setAutoCommit(final boolean autoCommit) throws SQLException {
 this.autoCommit = autoCommit;
 if (getConnections().isEmpty()) { // 無資料連線時，記錄方法呼叫
       recordMethodInvocation(Connection.class, "setAutoCommit", new Class[] {boolean.class}, new Object[] {autoCommit});
 return;
 }
 for (Connection each : getConnections()) {
       each.setAutoCommit(autoCommit);
 }
}

JdbcMethodInvocation，反射呼叫JDBC相關方法的工具類：

public class JdbcMethodInvocation {

 /**
    * 方法
    */
 @Getter
 private final Method method;
 /**
    * 方法引數
    */
 @Getter
 private final Object[] arguments;

 /**
    *  呼叫方法.
    * 
    * @param target 目標物件
    */
 public void invoke(final Object target) {
 try {
           method.invoke(target, arguments); // 反射呼叫
 } catch (final IllegalAccessException | InvocationTargetException ex) {
 throw new ShardingJdbcException("Invoke jdbc method exception", ex);
 }
 }
}

提供子類 #throwSQLExceptionIfNecessary() 丟擲異常鏈：

protected void throwSQLExceptionIfNecessary(final Collection<SQLException> exceptions) throws SQLException {
   if (exceptions.isEmpty()) { // 為空不丟擲異常
       return;
   }
   SQLException ex = new SQLException();
   for (SQLException each : exceptions) {
       ex.setNextException(each); // 異常鏈
   }
   throw ex;
}

3.2 AbstractDataSourceAdapter

AbstractDataSourceAdapter，資料來源適配類。

直接點選連結檢視原始碼。

3.3 AbstractConnectionAdapter

AbstractConnectionAdapter，資料庫連線適配類。

我們來瞅瞅大家最關心的事務相關方法的實現。

/**
* 是否自動提交
*/
private boolean autoCommit = true;

/**
* 獲得連結
*
* @return 連結
*/
protected abstract Collection<Connection> getConnections();

@Override
public final boolean getAutoCommit() throws SQLException {
   return autoCommit;
}

@Override
public final void setAutoCommit(final boolean autoCommit) throws SQLException {
   this.autoCommit = autoCommit;
   if (getConnections().isEmpty()) { // 無資料連線時，記錄方法呼叫
       recordMethodInvocation(Connection.class, "setAutoCommit", new Class[] {boolean.class}, new Object[] {autoCommit});
       return;
   }
   for (Connection each : getConnections()) {
       each.setAutoCommit(autoCommit);
   }
}

#setAutoCommit() 呼叫時，實際會設定其所持有的 Connection 的 autoCommit 屬性
#getConnections() 和分庫分表相關，因而僅抽象該方法，留給子類實現

@Override
public final void commit() throws SQLException {
   for (Connection each : getConnections()) {
       each.commit();
   }
}

@Override
public final void rollback() throws SQLException {
   Collection<SQLException> exceptions = new LinkedList<>();
   for (Connection each : getConnections()) {
       try {
           each.rollback();
       } catch (final SQLException ex) {
           exceptions.add(ex);
       }
   }
   throwSQLExceptionIfNecessary(exceptions);
}

#commit()、 #rollback() 呼叫時，實際呼叫其所持有的 Connection 的方法
異常情況下， #commit() 和 #rollback() 處理方式不同，筆者暫時不知道答案，求證後會進行更新
- #commit() 處理方式需要改成和 #rollback() 一樣。程式碼如下：

@Override
public final void commit() throws SQLException {
 Collection<SQLException> exceptions = new LinkedList<>();
 for (Connection each : getConnections()) {
 try {
           each.commit();
 } catch (final SQLException ex) {
           exceptions.add(ex);
 }
 }
   throwSQLExceptionIfNecessary(exceptions);
}

事務級別和是否只讀相關程式碼如下：

/**
* 只讀
*/
private boolean readOnly = true;
/**
* 事務級別
*/
private int transactionIsolation = TRANSACTION_READ_UNCOMMITTED;

@Override
public final void setReadOnly(final boolean readOnly) throws SQLException {
   this.readOnly = readOnly;
   if (getConnections().isEmpty()) {
       recordMethodInvocation(Connection.class, "setReadOnly", new Class[] {boolean.class}, new Object[] {readOnly});
       return;
   }
   for (Connection each : getConnections()) {
       each.setReadOnly(readOnly);
   }
}

@Override
public final void setTransactionIsolation(final int level) throws SQLException {
   transactionIsolation = level;
   if (getConnections().isEmpty()) {
       recordMethodInvocation(Connection.class, "setTransactionIsolation", new Class[] {int.class}, new Object[] {level});
       return;
   }
   for (Connection each : getConnections()) {
       each.setTransactionIsolation(level);
   }
}

3.4 AbstractStatementAdapter

AbstractStatementAdapter，靜態語句物件適配類。

@Override
public final int getUpdateCount() throws SQLException {
   long result = 0;
   boolean hasResult = false;
   for (Statement each : getRoutedStatements()) {
       if (each.getUpdateCount() > -1) {
           hasResult = true;
       }
       result += each.getUpdateCount();
   }
   if (result > Integer.MAX_VALUE) {
       result = Integer.MAX_VALUE;
   }
   return hasResult ? Long.valueOf(result).intValue() : -1;
}

/**
* 獲取路由的靜態語句物件集合.
* 
* @return 路由的靜態語句物件集合
*/
protected abstract Collection<? extends Statement> getRoutedStatements();

#getUpdateCount() 呼叫持有的 Statement 計算更新數量
#getRoutedStatements() 和分庫分表相關，因而僅抽象該方法，留給子類實現

3.5 AbstractPreparedStatementAdapter

AbstractPreparedStatementAdapter，預編譯語句物件的適配類。

#recordSetParameter()實現對佔位符引數的設定：

/**
* 記錄的設定引數方法陣列
*/
private final List<SetParameterMethodInvocation> setParameterMethodInvocations = new LinkedList<>();
/**
* 引數
*/
@Getter
private final List<Object> parameters = new ArrayList<>();

@Override
public final void setInt(final int parameterIndex, final int x) throws SQLException {
   setParameter(parameterIndex, x);
   recordSetParameter("setInt", new Class[]{int.class, int.class}, parameterIndex, x);
}

/**
* 記錄佔位符引數
*
* @param parameterIndex 佔位符引數位置
* @param value 引數
*/
private void setParameter(final int parameterIndex, final Object value) {
   if (parameters.size() == parameterIndex - 1) {
       parameters.add(value);
       return;
   }
   for (int i = parameters.size(); i <= parameterIndex - 1; i++) { // 用 null 填充前面未設定的位置
       parameters.add(null);
   }
   parameters.set(parameterIndex - 1, value);
}

/**
* 記錄設定引數方法呼叫
*
* @param methodName 方法名，例如 setInt、setLong 等
* @param argumentTypes 引數型別
* @param arguments 引數
*/
private void recordSetParameter(final String methodName, final Class[] argumentTypes, final Object... arguments) {
   try {
       setParameterMethodInvocations.add(new SetParameterMethodInvocation(PreparedStatement.class.getMethod(methodName, argumentTypes), arguments, arguments[1]));
   } catch (final NoSuchMethodException ex) {
       throw new ShardingJdbcException(ex);
   }
}

/**
* 回放記錄的設定引數方法呼叫
*
* @param preparedStatement 預編譯語句物件
*/
protected void replaySetParameter(final PreparedStatement preparedStatement) {
   addParameters();
   for (SetParameterMethodInvocation each : setParameterMethodInvocations) {
       updateParameterValues(each, parameters.get(each.getIndex() - 1)); // 同一個位置多次設定，值可能不一樣，需要更新下
       each.invoke(preparedStatement);
   }
}

/**
* 當使用分散式主鍵時，生成後會新增到 parameters，此時 parameters 數量多於 setParameterMethodInvocations，需要生成該分散式主鍵的 SetParameterMethodInvocation
*/
private void addParameters() {
   for (int i = setParameterMethodInvocations.size(); i < parameters.size(); i++) {
       recordSetParameter("setObject", new Class[]{int.class, Object.class}, i + 1, parameters.get(i));
   }
}

private void updateParameterValues(final SetParameterMethodInvocation setParameterMethodInvocation, final Object value) {
   if (!Objects.equals(setParameterMethodInvocation.getValue(), value)) {
       setParameterMethodInvocation.changeValueArgument(value); // 修改佔位符引數
   }
}

邏輯類似 WrapperAdapter 的 #recordMethodInvocation()， #replayMethodsInvocation()，請認真閱讀程式碼註釋
SetParameterMethodInvocation，繼承 JdbcMethodInvocation，反射呼叫引數設定方法的工具類：

public final class SetParameterMethodInvocation extends JdbcMethodInvocation {
/**


 * 位置


 */


@Getter


private final int index;


/**


 * 引數值


 */


@Getter


private final Object value;



/**


 * 設定引數值.


 * 


 * @param value 引數值


 */


public void changeValueArgument(final Object value) {


    getArguments()[1] = value;


}

}

3.6 AbstractResultSetAdapter

AbstractResultSetAdapter，代理結果集介面卡。

public abstract class AbstractResultSetAdapter extends AbstractUnsupportedOperationResultSet {
    /**
     * 結果集集合
     */
    @Getter
    private final List<ResultSet> resultSets;

    @Override
    // TODO should return sharding statement in future
    public final Statement getStatement() throws SQLException {
        return getResultSets().get(0).getStatement();
    }

    @Override
    public final ResultSetMetaData getMetaData() throws SQLException {
        return getResultSets().get(0).getMetaData();
    }

    @Override
    public int findColumn(final String columnLabel) throws SQLException {
        return getResultSets().get(0).findColumn(columnLabel);
    }

    // .... 省略其它方法
}

4. 插入流程

插入使用分散式主鍵例子程式碼如下：

// 程式碼僅僅是例子，生產環境下請注意異常處理和資源關閉
String sql = "INSERT INTO t_order(uid, nickname, pid) VALUES (1, '2', ?)";
DataSource dataSource = new ShardingDataSource(shardingRule);
Connection conn = dataSource.getConnection();
PreparedStatement ps = conn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS); // 返回主鍵需要  Statement.RETURN_GENERATED_KEYS
ps.setLong(1, 100);
ps.executeUpdate();
ResultSet rs = ps.getGeneratedKeys();
if (rs.next()) {
    System.out.println("id:" + rs.getLong(1));
}

呼叫 #executeUpdate() 方法，內部過程如下：

是不是對上層完全透明？！我們來看看內部是怎麼實現的。

// ShardingPreparedStatement.java
@Override
public int executeUpdate() throws SQLException {
   try {
       Collection<PreparedStatementUnit> preparedStatementUnits = route();
       return new PreparedStatementExecutor(
               getShardingConnection().getShardingContext().getExecutorEngine(), getRouteResult().getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeUpdate();
   } finally {
       clearBatch();
   }
}

#route() 分庫分表路由，獲得預編譯語句物件執行單元( PreparedStatementUnit )集合。
1. public final class PreparedStatementUnit implements BaseStatementUnit {
2. /**
3. * SQL 執行單元
4. */
5. private final SQLExecutionUnit sqlExecutionUnit;
6. /**
7. * 預編譯語句物件
8. */
9. private final PreparedStatement statement;
10. }
#executeUpdate() 呼叫執行引擎並行執行多個預編譯語句物件。執行時，最終呼叫預編譯語句物件( PreparedStatement )。我們來看一個例子：
1. // PreparedStatementExecutor.java
2. public int executeUpdate() {
3. Context context = MetricsContext.start("ShardingPreparedStatement-executeUpdate");
4. try {
5. List<Integer> results = executorEngine.executePreparedStatement(sqlType, preparedStatementUnits, parameters, new ExecuteCallback<Integer>() { @Override
6. public Integer execute(final BaseStatementUnit baseStatementUnit) throws Exception {
7. // 呼叫 PreparedStatement#executeUpdate()
8. return ((PreparedStatement) baseStatementUnit.getStatement()).executeUpdate();
9. }
10. });
11. return accumulate(results);
12. } finally {
13. MetricsContext.stop(context);
14. }
15. }

// ShardingPreparedStatement.java
private Collection<PreparedStatementUnit> route() throws SQLException {
   Collection<PreparedStatementUnit> result = new LinkedList<>();
   // 路由
   setRouteResult(routingEngine.route(getParameters()));
   // 遍歷 SQL 執行單元
   for (SQLExecutionUnit each : getRouteResult().getExecutionUnits()) {
       SQLType sqlType = getRouteResult().getSqlStatement().getType();
       Collection<PreparedStatement> preparedStatements;
       // 建立實際的 PreparedStatement
       if (SQLType.DDL == sqlType) {
           preparedStatements = generatePreparedStatementForDDL(each);
       } else {
           preparedStatements = Collections.singletonList(generatePreparedStatement(each));
       }
       getRoutedStatements().addAll(preparedStatements);
       // 回放設定佔位符引數到 PreparedStatement
       for (PreparedStatement preparedStatement : preparedStatements) {
           replaySetParameter(preparedStatement);
           result.add(new PreparedStatementUnit(each, preparedStatement));
       }
   }
   return result;
}

/**
* 建立 PreparedStatement
*
* @param sqlExecutionUnit SQL 執行單元
* @return PreparedStatement
* @throws SQLException 當 JDBC 操作發生異常時
*/
private PreparedStatement generatePreparedStatement(final SQLExecutionUnit sqlExecutionUnit) throws SQLException {
   Optional<GeneratedKey> generatedKey = getGeneratedKey();
   // 獲得連線
   Connection connection = getShardingConnection().getConnection(sqlExecutionUnit.getDataSource(), getRouteResult().getSqlStatement().getType());
   // 宣告返回主鍵
   if (isReturnGeneratedKeys() || isReturnGeneratedKeys() && generatedKey.isPresent()) {
       return connection.prepareStatement(sqlExecutionUnit.getSql(), RETURN_GENERATED_KEYS);
   }
   return connection.prepareStatement(sqlExecutionUnit.getSql(), getResultSetType(), getResultSetConcurrency(), getResultSetHoldability());
}

呼叫 #generatePreparedStatement() 建立 PreparedStatement，後呼叫 #replaySetParameter() 回放設定佔位符引數到 PreparedStatement
當 宣告返回主鍵 時，即 #isReturnGeneratedKeys() 返回 true 時，呼叫 connection.prepareStatement(sqlExecutionUnit.getSql(),RETURN_GENERATED_KEYS)。為什麼該方法會返回 true？上文例子 conn.prepareStatement(sql,Statement.RETURN_GENERATED_KEYS) 宣告返回主鍵後，插入執行完成，我們呼叫 #getGeneratedKeys() 可以獲得主鍵：
1. // ShardingStatement.java
2. @Override
3. public ResultSet getGeneratedKeys() throws SQLException {
4. Optional<GeneratedKey> generatedKey = getGeneratedKey();
5. // 分散式主鍵
6. if (generatedKey.isPresent() && returnGeneratedKeys) {
7. return new GeneratedKeysResultSet(routeResult.getGeneratedKeys().iterator(), generatedKey.get().getColumn(), this);
8. }
9. // 資料庫自增
10. if (1 == getRoutedStatements().size()) {
11. return getRoutedStatements().iterator().next().getGeneratedKeys();
12. }
13. return new GeneratedKeysResultSet();
14. }
15. // ShardingConnection.java
16. @Override
17. public PreparedStatement prepareStatement(final String sql, final String[] columnNames) throws SQLException {
18. return new ShardingPreparedStatement(this, sql, Statement.RETURN_GENERATED_KEYS);
19. }
21. // ShardingPreparedStatement.java
22. public ShardingPreparedStatement(final ShardingConnection shardingConnection, final String sql, final int autoGeneratedKeys) {
23. this(shardingConnection, sql);
24. if (RETURN_GENERATED_KEYS == autoGeneratedKeys) {
25. markReturnGeneratedKeys();
26. }
27. }
28. protected final void markReturnGeneratedKeys() {
29. returnGeneratedKeys = true;
30. }
呼叫 ShardingConnection#getConnection() 方法獲得該 PreparedStatement 對應的真實資料庫連線( Connection )：
- 呼叫 #getCachedConnection() 嘗試獲得已快取的資料庫連線；如果快取中不存在，獲取到連線後會進行快取
- 從 ShardingRule 配置的 DataSourceRule 獲取真實的資料來源( DataSource )
- MasterSlaveDataSource 實現主從資料來源封裝，我們在下小節分享
- 呼叫 #replayMethodsInvocation() 回放記錄的 Connection 方法
1. // ShardingConnection.java
2. /**
3. * 根據資料來源名稱獲取相應的資料庫連線.
4. *
5. * @param dataSourceName 資料來源名稱
6. * @param sqlType SQL語句型別
7. * @return 資料庫連線
8. * @throws SQLException SQL異常
9. */
10. public Connection getConnection(final String dataSourceName, final SQLType sqlType) throws SQLException {
11. // 從連線快取中獲取連線
12. Optional<Connection> connection = getCachedConnection(dataSourceName, sqlType);
13. if (connection.isPresent()) {
14. return connection.get();
15. }
16. Context metricsContext = MetricsContext.start(Joiner.on("-").join("ShardingConnection-getConnection", dataSourceName));
17. //
18. DataSource dataSource = shardingContext.getShardingRule().getDataSourceRule().getDataSource(dataSourceName);
19. Preconditions.checkState(null != dataSource, "Missing the rule of %s in DataSourceRule", dataSourceName);
20. String realDataSourceName;
21. if (dataSource instanceof MasterSlaveDataSource) {
22. dataSource = ((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
23. realDataSourceName = MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
24. } else {
25. realDataSourceName = dataSourceName;
26. }
27. Connection result = dataSource.getConnection();
28. MetricsContext.stop(metricsContext);
29. // 新增到連線快取
30. connectionMap.put(realDataSourceName, result);
31. // 回放 Connection 方法
32. replayMethodsInvocation(result);
33. return result;
34. }
36. private Optional<Connection> getCachedConnection(final String dataSourceName, final SQLType sqlType) {
37. String key = connectionMap.containsKey(dataSourceName) ? dataSourceName : MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
38. return Optional.fromNullable(connectionMap.get(key));
39. }

插入實現的程式碼基本分享完了，因為是不斷程式碼下鑽的方式分析，可以反向向上在理理，會更加清晰。

5. 查詢流程

單純從 core 包裡的 JDBC 實現，查詢流程 #executeQuery() 和 #execute() 基本一致，差別在於執行和多結果集歸併。

@Override
public ResultSet executeQuery() throws SQLException {
   ResultSet result;
   try {
       // 路由
       Collection<PreparedStatementUnit> preparedStatementUnits = route();
       // 執行
       List<ResultSet> resultSets = new PreparedStatementExecutor(
               getShardingConnection().getShardingContext().getExecutorEngine(), getRouteResult().getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeQuery();
       // 結果歸併
       result = new ShardingResultSet(resultSets, new MergeEngine(
               getShardingConnection().getShardingContext().getDatabaseType(), resultSets, (SelectStatement) getRouteResult().getSqlStatement()).merge());
   } finally {
       clearBatch();
   }
   // 設定結果集
   setCurrentResultSet(result);
   return result;
}

SQL執行 感興趣的同學可以看：《Sharding-JDBC 原始碼分析 —— SQL 執行》
結果歸併 感興趣的同學可以看：《Sharding-JDBC 原始碼分析 —— 結果歸併》
結果歸併 #merge() 完後，建立分片結果集( ShardingResultSet )

public final class ShardingResultSet extends AbstractResultSetAdapter {
 /**
     * 歸併結果集
     */
 private final ResultSetMerger mergeResultSet;
@Override


public int getInt(final int columnIndex) throws SQLException {


    Object result = mergeResultSet.getValue(columnIndex, int.class);


    wasNull = null == result;


    return (int) ResultSetUtil.convertValue(result, int.class);


}


 


@Override


public int getInt(final String columnLabel) throws SQLException {


    Object result = mergeResultSet.getValue(columnLabel, int.class);


    wasNull = null == result;


    return (int) ResultSetUtil.convertValue(result, int.class);


}


 


// .... 隱藏其他類似 getXXXX() 方法

}

6. 讀寫分離

建議前置閱讀：《官方文件 —— 讀寫分離》

當你有讀寫分離的需求時，將 ShardingRule 配置對應的資料來源 從 ShardingDataSource 替換成 MasterSlaveDataSource。我們來看看 MasterSlaveDataSource 的功能和實現。

支援一主多從的讀寫分離配置，可配合分庫分表使用

// MasterSlaveDataSourceFactory.java
public final class MasterSlaveDataSourceFactory {
    /**
     * 建立讀寫分離資料來源.
     * 
     * @param name 讀寫分離資料來源名稱
     * @param masterDataSource 主節點資料來源
     * @param slaveDataSource 從節點資料來源
     * @param otherSlaveDataSources 其他從節點資料來源
     * @return 讀寫分離資料來源
     */
    public static DataSource createDataSource(final String name, final DataSource masterDataSource, final DataSource slaveDataSource, final DataSource... otherSlaveDataSources) {
        return new MasterSlaveDataSource(name, masterDataSource, Lists.asList(slaveDataSource, otherSlaveDataSources));
    }
}

// MasterSlaveDataSource.java
public final class MasterSlaveDataSource extends AbstractDataSourceAdapter {
    /**
     * 資料來源名
     */
    private final String name;
    /**
     * 主資料來源
     */
    @Getter
    private final DataSource masterDataSource;
    /**
     * 從資料來源集合
     */
    @Getter
    private final List<DataSource> slaveDataSources;
}

同一執行緒且同一資料庫連線內，如有寫入操作，以後的讀操作均從主庫讀取，用於保證資料一致性。

// ShardingConnection.java
public Connection getConnection(final String dataSourceName, final SQLType sqlType) throws SQLException {
   // .... 省略部分程式碼
   String realDataSourceName;
   if (dataSource instanceof MasterSlaveDataSource) { // 讀寫分離
       dataSource = ((MasterSlaveDataSource) dataSource).getDataSource(sqlType);
       realDataSourceName = MasterSlaveDataSource.getDataSourceName(dataSourceName, sqlType);
   } else {
       realDataSourceName = dataSourceName;
   }
   Connection result = dataSource.getConnection();
   // .... 省略部分程式碼
}

// MasterSlaveDataSource.java
/**
* 當前執行緒是否是 DML 操作標識
*/
private static final ThreadLocal<Boolean> DML_FLAG = new ThreadLocal<Boolean>() {

   @Override
   protected Boolean initialValue() {
       return false;
   }
};
/**
* 從庫負載均衡策略
*/
private final SlaveLoadBalanceStrategy slaveLoadBalanceStrategy = new RoundRobinSlaveLoadBalanceStrategy();

/**
* 獲取主或從節點的資料來源.
*
* @param sqlType SQL型別
* @return 主或從節點的資料來源
*/
public DataSource getDataSource(final SQLType sqlType) {
   if (isMasterRoute(sqlType)) {
       DML_FLAG.set(true);
       return masterDataSource;
   }
   return slaveLoadBalanceStrategy.getDataSource(name, slaveDataSources);
}

private static boolean isMasterRoute(final SQLType sqlType) {
   return SQLType.DQL != sqlType || DML_FLAG.get() || HintManagerHolder.isMasterRouteOnly();
}

ShardingConnection 獲取到的資料來源是 MasterSlaveDataSource 時，呼叫 MasterSlaveDataSource#getConnection() 方法獲取真實的資料來源
通過 #isMasterRoute() 判斷是否讀取主庫，以下三種情況會訪問主庫：
- 非查詢語句 (DQL)
- 該資料來源在當前執行緒訪問過主庫：通過執行緒變數 DML_FLAG 實現
- 強制主庫：程式裡呼叫 HintManager.getInstance().setMasterRouteOnly() 實現
訪問從庫時，會通過負載均衡策略( SlaveLoadBalanceStrategy ) 選擇一個從庫
- MasterSlaveDataSource 預設使用 RoundRobinSlaveLoadBalanceStrategy，暫時不支援配置
- RoundRobinSlaveLoadBalanceStrategy，輪詢負載均衡策略，每個從節點訪問次數均衡，暫不支援資料來源故障移除

// SlaveLoadBalanceStrategy.java
public interface SlaveLoadBalanceStrategy {
/**


 * 根據負載均衡策略獲取從庫資料來源.


 * 


 * @param name 讀寫分離資料來源名稱


 * @param slaveDataSources 從庫資料來源列表


 * @return 選中的從庫資料來源


 */


DataSource getDataSource(String name, List&lt;DataSource&gt; slaveDataSources);

}

// RoundRobinSlaveLoadBalanceStrategy.java
public final class RoundRobinSlaveLoadBalanceStrategy implements SlaveLoadBalanceStrategy {
private static final ConcurrentHashMap&lt;String, AtomicInteger&gt; COUNT_MAP = new ConcurrentHashMap&lt;&gt;();


 


@Override


public DataSource getDataSource(final String name, final List&lt;DataSource&gt; slaveDataSources) {


    AtomicInteger count = COUNT_MAP.containsKey(name) ? COUNT_MAP.get(name) : new AtomicInteger(0);


    COUNT_MAP.putIfAbsent(name, count);


    count.compareAndSet(slaveDataSources.size(), 0);


    return slaveDataSources.get(count.getAndIncrement() % slaveDataSources.size());


}

}

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— JDBC實現與讀寫分離

1. 概述

2. unspported 包

3. adapter 包

3.1 WrapperAdapter

3.2 AbstractDataSourceAdapter

3.3 AbstractConnectionAdapter

3.4 AbstractStatementAdapter

3.5 AbstractPreparedStatementAdapter

3.6 AbstractResultSetAdapter

4. 插入流程

5. 查詢流程

6. 讀寫分離

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— JDBC實現與讀寫分離

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— SQL 解析（三）之查詢SQL

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— SQL 解析（一）之語法解析

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— 結果歸併

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— 事務（一）之BED

mysql中介軟體proxysql實現mysql讀寫分離

原始碼分析：ReentrantReadWriteLock之讀寫鎖

MySQL資料庫的主從同步配置與讀寫分離

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 路由（一）之分庫分表配置

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 解析（六）之刪除SQL

資料庫[分庫分表]中介軟體 Sharding-JDBC 原始碼分析 —— SQL 解析（五）之更新SQL解析

資料庫[分庫分表]中介軟體 Sharding-JDBC 原始碼分析 —— SQL 解析（四）之插入SQL

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 執行

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— 分散式主鍵

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 改寫

資料庫分庫分表中介軟體 Sharding-JDBC 原始碼分析 —— SQL 路由（二）之分庫分表路由

分散式資料庫中介軟體、產品——sharding-jdbc、mycat、drds

一文快速入門分庫分表中介軟體 Sharding-JDBC （必修課）

資料庫中介軟體 MyCAT 原始碼分析 —— 【單庫單表】查詢

資料庫中介軟體 MyCAT原始碼分析：【單庫單表】插入

資料庫中介軟體 Sharding-JDBC 原始碼分析 —— JDBC實現與讀寫分離

1. 概述

2. unspported 包

3. adapter 包

3.1 WrapperAdapter

3.2 AbstractDataSourceAdapter

3.3 AbstractConnectionAdapter

3.4 AbstractStatementAdapter

3.5 AbstractPreparedStatementAdapter

3.6 AbstractResultSetAdapter

4. 插入流程

5. 查詢流程

6. 讀寫分離

相關推薦