歡迎光臨
每天分享高質量文章
當前位置：知識星球 > 後端 > 正文

【死磕Sharding-jdbc】—結果合併

2018-05-23 分類：後端 閱讀(235) 評論(0)

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

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

單表查詢之結果合併

接下來以執行 SELECT o.*FROM t_order o whereo.user_id=10order byo.order_id desc limit 2,3分析下麵這段Java程式碼是如何對結果進行合併的：

  1. result = new ShardingResultSet(resultSets, new MergeEngine(resultSets, (SelectStatement) routeResult.getSqlStatement()).merge());code> 
  3. MergeEngine.merge()方法的原始碼如下：
  5. <code class="java">public ResultSetMerger merge() throws SQLException {
  6.     selectStatement.setIndexForItems(columnLabelIndexMap);
  7.     return decorate(build());
  8. }

build()方法原始碼如下:

  1. private ResultSetMerger build() throws SQLException {
  2.     // 說明：GroupBy***ResultSetMerger在第六篇文章單獨講解，所以此次分析的SQL條件中沒有group by
  3.     if (!selectStatement.getGroupByItems().isEmpty() || !selectStatement.getAggregationSelectItems().isEmpty()) {
  4.         if (selectStatement.isSameGroupByAndOrderByItems()) {
  5.             return new GroupByStreamResultSetMerger(columnLabelIndexMap, resultSets, selectStatement);
  6.         } else {
  7.             return new GroupByMemoryResultSetMerger(columnLabelIndexMap, resultSets, selectStatement);
  8.         }
  9.     }
  10.     // 如果select陳述句中有order by欄位，那麼需要OrderByStreamResultSetMerger對結果處理
  11.     if (!selectStatement.getOrderByItems().isEmpty()) {
  12.         return new OrderByStreamResultSetMerger(resultSets, selectStatement.getOrderByItems());
  13.     }
  14.     return new IteratorStreamResultSetMerger(resultSets);
  15. }

根據這段程式碼可知，其作用是根據sql陳述句選擇多個不同的ResultSetMerger對結果進行合併處理，ResultSetMerger實現有這幾種：GroupByStreamResultSetMerger，GroupByMemoryResultSetMerger，OrderByStreamResultSetMerger，IteratorStreamResultSetMerger，LimitDecoratorResultSetMerger；以測試SQL SELECT o.*FROM t_order o whereo.user_id=10order byo.order_id desc limit 2,3為例，沒有group by，但是有order by，所以使用到了OrderByStreamResultSetMerger和LimitDecoratorResultSetMerger對結果進行合併（GroupByStreamResultSetMerger&GroupByMemoryResultSetMerger;後面單獨講解）

decorate()原始碼如下：

  1. private ResultSetMerger decorate(final ResultSetMerger resultSetMerger) throws SQLException {
  2.     ResultSetMerger result = resultSetMerger;
  3.     // 如果SQL陳述句中有limist，還需要LimitDecoratorResultSetMerger配合進行結果歸併；
  4.     if (null != selectStatement.getLimit()) {
  5.         result = new LimitDecoratorResultSetMerger(result, selectStatement.getLimit());
  6.     }
  7.     return result;
  8. }

接下來將以執行SQL： SELECT o.*FROM t_order o whereo.user_id=10order byo.order_id desc limit 2,3（該SQL會被改寫成 SELECT o.*,o.order_id AS ORDER_BY_DERIVED_0 FROM t_order_0 o whereo.user_id=?order byo.order_id desc limit 2,3）為例，一一講解OrderByStreamResultSetMerger，LimitDecoratorResultSetMerger和IteratorStreamResultSetMerger，瞭解這幾個ResultSetMerger的原理；

OrderByStreamResultSetMerger

OrderByStreamResultSetMerger的核心原始碼如下：

  1. private final Queue<OrderByValue> orderByValuesQueue;
  3. public OrderByStreamResultSetMerger(final List<ResultSet> resultSets, final List<OrderItem> orderByItems) throws SQLException {
  4.     // sql中order by列的資訊，實體sql是order by order_id desc，即此處就是order_id
  5.     this.orderByItems = orderByItems;
  6.     // 初始化一個優先順序佇列，優先順序佇列中的元素會根據OrderByValue中compareTo()方法排序，並且SQL重寫後傳送到多少個標的實際表，List的size就有多大，Queue的capacity就有多大；
  7.     this.orderByValuesQueue = new PriorityQueue<>(resultSets.size());
  8.     // 將結果壓入佇列中
  9.     orderResultSetsToQueue(resultSets);
  10.     isFirstNext = true;
  11. }
  13. private void orderResultSetsToQueue(final List<ResultSet> resultSets) throws SQLException {
  14.     // 遍歷resultSets--在多少個標的實際表上執行SQL，該集合的size就有多大
  15.     for (ResultSet each : resultSets) {
  16.         // 將ResultSet和排序列資訊封裝成一個OrderByValue型別
  17.         OrderByValue orderByValue = new OrderByValue(each, orderByItems);
  18.         // 如果值存在，那麼壓入佇列中
  19.         if (orderByValue.next()) {
  20.             orderByValuesQueue.offer(orderByValue);
  21.         }
  22.     }
  23.     // 重置currentResultSet的位置：如果佇列不為空，那麼將佇列的頂部(peek)位置設定為currentResultSet的位置
  24.     setCurrentResultSet(orderByValuesQueue.isEmpty() ? resultSets.get(0) : orderByValuesQueue.peek().getResultSet());
  25. }
  27. @Override
  28. public boolean next() throws SQLException {
  29.     // 呼叫next()判斷是否還有值, 如果佇列為空, 表示沒有任何值, 那麼直接傳回false
  30.     if (orderByValuesQueue.isEmpty()) {
  31.         return false;
  32.     }
  33.     // 如果佇列不為空, 那麼第一次一定傳回true；即有結果可取（且將isFirstNext置為false，表示接下來的請求都不是第一次請求next()方法）
  34.     if (isFirstNext) {
  35.         isFirstNext = false;
  36.         return true;
  37.     }
  38.     // 從佇列中彈出第一個元素（因為是優先順序佇列，所以poll()傳回的值，就是此次要取的值）
  39.     OrderByValue firstOrderByValue = orderByValuesQueue.poll();
  40.     // 如果它的next()存在，那麼將它的next()再新增到佇列中
  41.     if (firstOrderByValue.next()) {
  42.         orderByValuesQueue.offer(firstOrderByValue);
  43.     }
  44.     // 佇列中所有元素全部處理完後就傳回false
  45.     if (orderByValuesQueue.isEmpty()) {
  46.         return false;
  47.     }
  48.     // 再次重置currentResultSet的位置為佇列的頂部位置；
  49.     setCurrentResultSet(orderByValuesQueue.peek().getResultSet());
  50.     return true;
  51. }

繼續深入剖析：這段程式碼初看可能有點繞，假設執行SQL SELECT o.*FROM t_order o whereo.user_id=10order byo.order_id desc limit 3會分發到兩個標的實際表，且第一個實際表傳回的結果是1，3，5，7，9；第二個實際表傳回的結果是2，4，6，8，10；那麼，經過OrderByStreamResultSetMerger的構造方法中的orderResultSetsToQueue()方法後， Queue<OrderByValue>orderByValuesQueue中包含兩個OrderByValue，一個是10，一個是9；接下來取值執行過程如下：

  1. 取得10，並且10的next()是8，然後執行orderByValuesQueue.offer(8);，這時候orderByValuesQueue中包含8和9；

  2. 取得9，並且9的next()是7，然後執行orderByValuesQueue.offer(7);，這時候orderByValuesQueue中包含7和8；

  3. 取得8，並且8的next()是6，然後執行orderByValuesQueue.offer(6);，這時候orderByValuesQueue中包含7和6； 取值數量已經達到limit 3的限制（原始碼在LimitDecoratorResultSetMerger中的next()方法中），退出；

這段程式碼執行示意圖如下所示：

LimitDecoratorResultSetMerger

LimitDecoratorResultSetMerger核心原始碼如下：

  1. public LimitDecoratorResultSetMerger(final ResultSetMerger resultSetMerger, final Limit limit) throws SQLException {
  2.     super(resultSetMerger);
  3.     // limit賦值（Limit物件包括limit m,n中的m和n兩個值）
  4.     this.limit = limit;
  5.     // 判斷是否會跳過所有的結果項，即判斷是否有符合條件的結果
  6.     skipAll = skipOffset();
  7. }
  9. private boolean skipOffset() throws SQLException {
  10.     // 假定limit.getOffsetValue()就是offset，實體sql中為limit 2,3，所以offset=2
  11.     for (int i = 0; i < limit.getOffsetValue(); i++) {
  12.         // 嘗試從OrderByStreamResultSetMerger生成的優先順序佇列中跳過offset個元素，如果.next()一直為true，表示有足夠符合條件的結果，那麼傳回false；否則沒有足夠符合條件的結果，那麼傳回true；即skilAll=true就表示跳過了所有沒有符合條件的結果；
  13.         if (!getResultSetMerger().next()) {
  14.             return true;
  15.         }
  16.     }
  17.     // limit m,n的sql會被重寫為limit 0, m+n，所以limit.isRowCountRewriteFlag()為true，rowNumber的值為0；
  18.     rowNumber = limit.isRowCountRewriteFlag() ? 0 : limit.getOffsetValue();
  19.     return false;
  20. }
  22. @Override
  23. public boolean next() throws SQLException {
  24.     // 如果skipAll為true，即跳過所有，表示沒有任何符合條件的值，那麼傳回false
  25.     if (skipAll) {
  26.         return false;
  27.     }
  28.     if (limit.getRowCountValue() > -1) {
  29.         // 每次next()獲取值後，rowNumber自增，當自增rowCountValue次後，就不能再往下繼續取值了，因為條件limit 2,3（rowCountValue=3）限制了
  30.         return ++rowNumber <= limit.getRowCountValue() && getResultSetMerger().next();
  31.     }
  32.     return getResultSetMerger().next();
  33. }

IteratorStreamResultSetMerger

構造方法核心原始碼：

  1. private final Iterator<ResultSet> resultSets;
  3. public IteratorStreamResultSetMerger(final List<ResultSet> resultSets) {
  4.     // 將List改成Iterator，方便接下來迭代取得結果；
  5.     this.resultSets = resultSets.iterator();
  6.     // 重置currentResultSet
  7.     setCurrentResultSet(this.resultSets.next());
  8. }

END

贊(0)
標籤：

相關推薦

熱門標籤

iOS (11238)微軟 (4955)Linux (4274)安全 (4180)Python (4161)效能 (3165)運維 (2774)最佳化 (2419).NET (2262)Google (2136)機器學習 (1795)併發 (1613)分散式 (1559)叢集 (1240)SQL (1174)Mysql (1060)區塊鏈 (1017)Docker (977)微服務 (922)面試 (919)Apache (743)NLP (725)Redis (719)Android (668)Git (640)架構師 (632)Nginx (630)Facebook (599)JVM (595)爬蟲 (476)

熱門文章

分享創造快樂

© 2024 知識星球   網站地圖