點選上方“Java技術驛站”，選擇“置頂公眾號”。

有內涵、有價值的文章第一時間送達！

繼續以 sharding-jdbc-example-jdbc模組中的 com.dangdang.ddframe.rdb.sharding.example.jdbc.Main為基礎，剖析分庫分表簡單查詢SQL實現– printSimpleSelect(dataSource);，即如何執行簡單的查詢SQL，接下來的分析以執行SQL陳述句 "SELECT o.* FROM t_order o where o.user_id=? AND o.order_id=?"為例；

單表查詢

Main中 printSimpleSelect()方法呼叫 preparedStatement.executeQuery()，即呼叫ShardingPreparedStatement中的 executeQuery()方法，核心原始碼如下：

@Override
public ResultSet executeQuery() throws SQLException {
    ResultSet result;
    try {
        // 核心方法route()，即解析SQL如何路由執行
        Collection<PreparedStatementUnit> preparedStatementUnits = route();
        // 根據路由資訊執行SQL
        List<ResultSet> resultSets = new PreparedStatementExecutor(
                getConnection().getShardingContext().getExecutorEngine(), routeResult.getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeQuery();
        // 對傳回的結果進行merge合併
        result = new ShardingResultSet(resultSets, new MergeEngine(resultSets, (SelectStatement) routeResult.getSqlStatement()).merge());
    } finally {
        clearBatch();
    }
    currentResultSet = result;
    return result;
}

透過上面的原始碼可知，SQL查詢兩個核心：路由和結果合併，接下來一一分析sharding-jdbc如何實現；

單表查詢之路由

接下來分析下麵這段程式碼是如何取得路由資訊的：

Collection<PreparedStatementUnit> preparedStatementUnits = route();

route()核心原始碼如下：

private Collection<PreparedStatementUnit> route() throws SQLException {
    Collection<PreparedStatementUnit> result = new LinkedList<>();
    // 呼叫PreparedStatementRoutingEngine中的route()方法，route()方法呼叫sqlRouter.route(logicSQL, parameters, sqlStatement)
    routeResult = routingEngine.route(getParameters());
    for (SQLExecutionUnit each : routeResult.getExecutionUnits()) {
        SQLType sqlType = routeResult.getSqlStatement().getType();
        Collection<PreparedStatement> preparedStatements;
        if (SQLType.DDL == sqlType) {
            preparedStatements = generatePreparedStatementForDDL(each);
        } else {
            preparedStatements = Collections.singletonList(generatePreparedStatement(each));
        }
        routedStatements.addAll(preparedStatements);
        for (PreparedStatement preparedStatement : preparedStatements) {
            replaySetParameter(preparedStatement);
            result.add(new PreparedStatementUnit(each, preparedStatement));
        }
    }
    return result;
}

SQLRouter介面有兩個實現類：DatabaseHintSQLRouter和ParsingSQLRouter，由於這裡沒有用hint語法強制執行某個庫，所以呼叫ParsingSQLRouter中的route()方法：

private RoutingResult route(final List<Object> parameters, final SQLStatement sqlStatement) {
    Collection<String> tableNames = sqlStatement.getTables().getTableNames();
    RoutingEngine routingEngine;
    // 如果sql中只有一個表名，或者多個表名之間是系結表關係，或者所有表都在預設資料源指定的資料庫中（即不參與分庫分表的表），那麼用SimpleRoutingEngine作為路由判斷引擎；
    if (1 == tableNames.size() || shardingRule.isAllBindingTables(tableNames) || shardingRule.isAllInDefaultDataSource(tableNames)) {
        routingEngine = new SimpleRoutingEngine(shardingRule, parameters, tableNames.iterator().next(), sqlStatement);
    } else {
        routingEngine = new ComplexRoutingEngine(shardingRule, parameters, tableNames, sqlStatement);
    }
    return routingEngine.route();
}

接下來分析一下SimpleRoutingEngine和ComplexRoutingEngine；

SimpleRoutingEngine

執行SQL： "SELECT o.* FROM t_order o where o.user_id=? AND o.order_id=?"時，由於SQL中只有一個表（1 == tableNames.size()），所以路由引擎是SimpleRoutingEngine； SimpleRoutingEngine.route()原始碼如下：

@Override
public RoutingResult route() {
    // 根據邏輯表得到tableRule，邏輯表為t_order；表規則的配置為：.actualTables(Arrays.asList("t_order_0", "t_order_1"))，所以有兩個實際表；
    TableRule tableRule = shardingRule.getTableRule(logicTableName);
    // 根據規則先路由資料源：即根據user_id取模路由
    Collection<String> routedDataSources = routeDataSources(tableRule);
    // routedMap儲存路由到的標的資料源和表的結果：key為資料源，value為該資料源下路由到的標的表集合
    Map<String, Collection<String>> routedMap = new LinkedHashMap<>(routedDataSources.size());
    // 遍歷路由到的標的資料源
    for (String each : routedDataSources) {
        // 再根據規則路由表：即根據order_id取模路由
        routedMap.put(each, routeTables(tableRule, each));
    }
    // 將得到的路由資料源和表資訊封裝到RoutingResult中，RoutingResult中有個TableUnits型別屬性，TableUnits類中有個List tableUnits屬性，TableUnit包含三個屬性：dataSourceName--資料源名稱，logicTableName--邏輯表名稱，actualTableName--實際表名稱，例如：TableUnit:{dataSourceName:ds_jdbc_1, logicTableName:t_order, actualTableName: t_order_1}
    return generateRoutingResult(tableRule, routedMap);
}

資料源路由詳細解讀：由於資料源的sharding策略為 databaseShardingStrategy(newDatabaseShardingStrategy("user_id",newModuloDatabaseShardingAlgorithm()))；且where條件為 whereo.user_id=?AND o.order_id=?，即where條件中有userid，根據取模路由策略，當userid為奇數時，資料源為dsjdbc1；當userid為偶數時，資料源為dsjdbc0； 表路由詳細解讀：表的sharding策略為 tableShardingStrategy(newTableShardingStrategy("order_id",newModuloTableShardingAlgorithm()))，即where條件中有orderid，根據取模路由策略，當orderid為奇數時，表為torder1；當orderid為偶數時，表為torder0； 綜上所述：最終需要執行的表數量為路由到的資料源個數路由到的實際表個數*；

實體1：where o.orderid=1001 AND o.userid=10，userid=10所以路由得到資料源為dsjdbc0; orderid=1001，路由得到實際表為torder1；那麼最終只需在dsjdbc0這個資料源中的torder1表中執行即可； 實體2：where o.orderid=1000，userid沒有值所以路由得到所有資料源dsjdbc0和dsjdbc1; orderid=1000，路由得到實際表為torder0；那麼最終需在dsjdbc0和dsjdbc1兩個資料源中的torder0表中執行即可； 實體3：where o.userid=11，userid=11所以路由得到資料源為dsjdbc1; orderid沒有值所以路由得到實際表為torder0和torder1；那麼最終只需在dsjdbc1這個資料源中的torder0和torder1表中執行即可；

ComplexRoutingEngine

待定… …

單表查詢之執行

路由完成後就決定了SQL需要在哪些資料源的哪些實際表中執行，接下來以執行 SELECT o.*FROM t_order o whereo.user_id=10為例分析下麵這段Java程式碼sharding-jdbc是如何執行的：

根據前面的路由分析可知，這條SQL會路由到dsjdbc0這個資料源中，且在所有實際表([torder0, torder1])中執行這個SQL；

List<ResultSet> resultSets = new PreparedStatementExecutor(
                getConnection().getShardingContext().getExecutorEngine(), routeResult.getSqlStatement().getType(), preparedStatementUnits, getParameters()).executeQuery();

執行的核心程式碼在ExecutorEngine中，核心原始碼如下：

public <T> List<T> executePreparedStatement(
        final SQLType sqlType, final Collection<PreparedStatementUnit> preparedStatementUnits, final List<Object> parameters, final ExecuteCallback<T> executeCallback) {
    // preparedStatementUnits就是前面路由分析結果：執行SQL select o.* from t_order o where o.user_id=10時，只需在ds_jdbc_0這個資料源中的t_order_0和t_order_1兩個實際表中執行即可；
    return execute(sqlType, preparedStatementUnits, Collections.singletonList(parameters), executeCallback);
}
private  <T> List<T> execute(
        final SQLType sqlType, final Collection extends BaseStatementUnit> baseStatementUnits, final List<List<Object>> parameterSets, final ExecuteCallback<T> executeCallback) {
    if (baseStatementUnits.isEmpty()) {
        return Collections.emptyList();
    }
    Iterator extends BaseStatementUnit> iterator = baseStatementUnits.iterator();
    // 第一個任務分離出來
    BaseStatementUnit firstInput = iterator.next();
    // 除第一個任務之外的任務非同步執行
    ListenableFuture<List<T>> restFutures = asyncExecute(sqlType, Lists.newArrayList(iterator), parameterSets, executeCallback);
    T firstOutput;
    List<T> restOutputs;
    try {
        // 第一個任務同步執行[猜測是不是考慮到分庫分表後只需路由到一個資料源中的一個表的SQL執行效能問題，最佳化這種SQL執行為同步執行？分庫分表後，面向使用者的API佔用了99%的請求量，而這些API對應的SQL 99%只需要在一個資料源上的一個實際表執行即可，例如根據訂單表根據user_id分庫分表後，查詢使用者的訂單資訊這種場景]
        firstOutput = syncExecute(sqlType, firstInput, parameterSets, executeCallback);
        // 取得其他非同步執行任務的結果
        restOutputs = restFutures.get();
        //CHECKSTYLE:OFF
    } catch (final Exception ex) {
        //CHECKSTYLE:ON
        ExecutorExceptionHandler.handleException(ex);
        return null;
    }
    List<T> result = Lists.newLinkedList(restOutputs);
    // 將第一個任務同步執行結果與其他任務非同步執行結果合併就是最終的結果
    result.add(0, firstOutput);
    return result;
}

非同步執行核心程式碼：

private final ListeningExecutorService executorService;
public ExecutorEngine(final int executorSize) {
    // 非同步執行的執行緒池是透過google-guava封裝的執行緒池，設定了執行緒名稱為增加了ShardingJDBC-***，增加了shutdown hook--應用關閉時最多等待60秒直到所有任務完成，從而實現優雅停機
    executorService = MoreExecutors.listeningDecorator(new ThreadPoolExecutor(
            executorSize, executorSize, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactoryBuilder().setDaemon(true).setNameFormat("ShardingJDBC-%d").build()));
    MoreExecutors.addDelayedShutdownHook(executorService, 60, TimeUnit.SECONDS);
}
private <T> ListenableFuture<List<T>> asyncExecute(
        final SQLType sqlType, final Collection<BaseStatementUnit> baseStatementUnits, final List<List<Object>> parameterSets, final ExecuteCallback<T> executeCallback) {
    // 構造一個存放非同步執行後的結果的list
    List<ListenableFuture<T>> result = new ArrayList<>(baseStatementUnits.size());
    final boolean isExceptionThrown = ExecutorExceptionHandler.isExceptionThrown();
    final Map<String, Object> dataMap = ExecutorDataMap.getDataMap();
    for (final BaseStatementUnit each : baseStatementUnits) {
        // 執行緒池方式非同步執行所有SQL，執行緒池在ExecutorEngine的構造方法中初始化；
        result.add(executorService.submit(new Callable<T>() {
            @Override
            public T call() throws Exception {
                return executeInternal(sqlType, each, parameterSets, executeCallback, isExceptionThrown, dataMap);
            }
        }));
    }
    // google-guava的方法--將所有非同步執行結果轉為list型別
    return Futures.allAsList(result);
}

同步執行核心程式碼：

private <T> T syncExecute(final SQLType sqlType, final BaseStatementUnit baseStatementUnit, final List<List<Object>> parameterSets, final ExecuteCallback<T> executeCallback) throws Exception {
        return executeInternal(sqlType, baseStatementUnit, parameterSets, executeCallback, ExecutorExceptionHandler.isExceptionThrown(), ExecutorDataMap.getDataMap());
    }

由同步執行核心程式碼和非同步執行核心程式碼可知，最終都是呼叫 executeInternal()，跟讀這個方法的原始碼可知：最終就是在標的資料庫表上執行 PreparedStatement的 execute***()方法；且在執行前會利用google-guava的EventBus釋出BEFOREEXECUTE的事件（執行完成後，如果執行成功還會釋出EXECUTESUCCESS事件，如果執行失敗釋出EXECUTE_FAILURE事件），部分核心原始碼如下：

// 釋出事件
List<AbstractExecutionEvent> events = new LinkedList<>();
if (parameterSets.isEmpty()) {
    // 構造無參SQL的事件（事件型別為BEFORE_EXECUTE）
    events.add(getExecutionEvent(sqlType, baseStatementUnit, Collections.emptyList()));
}
for (List<Object> each : parameterSets) {
    // 構造有參SQL的事件（事件型別為BEFORE_EXECUTE）
    events.add(getExecutionEvent(sqlType, baseStatementUnit, each));
}
// 呼叫google-guava的EventBus.post()提交事件
for (AbstractExecutionEvent event : events) {
    EventBusInstance.getInstance().post(event);
}
try {
    // 執行SQL
    result = executeCallback.execute(baseStatementUnit);
} catch (final SQLException ex) {
    // 如果執行過程中丟擲SQLException，即執行SQL失敗，那麼post一個EXECUTE_FAILURE型別的事件
    for (AbstractExecutionEvent each : events) {
        each.setEventExecutionType(EventExecutionType.EXECUTE_FAILURE);
        each.setException(Optional.of(ex));
        EventBusInstance.getInstance().post(each);
        ExecutorExceptionHandler.handleException(ex);
    }
    return null;
}
for (AbstractExecutionEvent each : events) {
    // // 如果執行成功，那麼post一個EXECUTE_SUCCESS型別的事件
    each.setEventExecutionType(EventExecutionType.EXECUTE_SUCCESS);
    EventBusInstance.getInstance().post(each);
}

接下來需要對並行執行後得到的結果集進行merge，下麵的sharding-jdbc原始碼分析系列文章繼續對其進行分析；

EventBus

說明：EventBus是google-guava提供的訊息釋出-訂閱類庫； google-guava的EventBus正確開啟姿勢：

1. 釋出事務：呼叫EventBus的post()--sharding-jdbc中釋出事務：EventBusInstance.getInstance().post(each)；

2. 訂閱事務：呼叫EventBus的register()--sharding-jdbc中註冊事務：EventBusInstance.getInstance().register(new BestEffortsDeliveryListener())；

EventBusInstance原始碼如下--EventBus全類名為 com.google.common.eventbus.EventBus：

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class EventBusInstance {
    private static final EventBus INSTANCE = new EventBus();
    /**
     * Get event bus instance.
     * 
     * @return event bus instance
     */
    public static EventBus getInstance() {
        return INSTANCE;
    }
}

END

阿飛哥

贊賞

長按二維碼向我轉賬

受蘋果公司新規定影響，微信 iOS 版的贊賞功能被關閉，可透過二維碼轉賬支援公眾號。

閱讀原文


微信掃一掃
使用小程式
即將開啟""小程式
取消
開啟