【死磕Sharding-jdbc】---结果合并

单表查询之结果合并

接下来以执行 SELECT o.*FROM t_order o whereo.user_id=10order byo.order_id desc limit 2,3分析下面这段Java代码是如何对结果进行合并的:

result = new ShardingResultSet(resultSets, new MergeEngine(resultSets, (SelectStatement) routeResult.getSqlStatement()).merge());</code> MergeEngine.merge()方法的源码如下:<code class="java">public ResultSetMerger merge() throws SQLException {    selectStatement.setIndexForItems(columnLabelIndexMap);    return decorate(build());}

build()方法源码如下:

private ResultSetMerger build() throws SQLException {    // 说明:GroupBy***ResultSetMerger在第六篇文章单独讲解,所以此次分析的SQL条件中没有group by    if (!selectStatement.getGroupByItems().isEmpty() || !selectStatement.getAggregationSelectItems().isEmpty()) {        if (selectStatement.isSameGroupByAndOrderByItems()) {            return new GroupByStreamResultSetMerger(columnLabelIndexMap, resultSets, selectStatement);        } else {            return new GroupByMemoryResultSetMerger(columnLabelIndexMap, resultSets, selectStatement);        }    }    // 如果select语句中有order by字段,那么需要OrderByStreamResultSetMerger对结果处理    if (!selectStatement.getOrderByItems().isEmpty()) {        return new OrderByStreamResultSetMerger(resultSets, selectStatement.getOrderByItems());    }    return new IteratorStreamResultSetMerger(resultSets);}

根据这段代码可知,其作用是根据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()源码如下:

private ResultSetMerger decorate(final ResultSetMerger resultSetMerger) throws SQLException {    ResultSetMerger result = resultSetMerger;    // 如果SQL语句中有limist,还需要LimitDecoratorResultSetMerger配合进行结果归并;    if (null != selectStatement.getLimit()) {        result = new LimitDecoratorResultSetMerger(result, selectStatement.getLimit());    }    return result;}

接下来将以执行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的核心源码如下:

private final Queue<OrderByValue> orderByValuesQueue;public OrderByStreamResultSetMerger(final List<ResultSet> resultSets, final List<OrderItem> orderByItems) throws SQLException {    // sql中order by列的信息,实例sql是order by order_id desc,即此处就是order_id    this.orderByItems = orderByItems;    // 初始化一个优先级队列,优先级队列中的元素会根据OrderByValue中compareTo()方法排序,并且SQL重写后发送到多少个目标实际表,List<ResultSet>的size就有多大,Queue的capacity就有多大;    this.orderByValuesQueue = new PriorityQueue<>(resultSets.size());    // 将结果压入队列中    orderResultSetsToQueue(resultSets);    isFirstNext = true;}private void orderResultSetsToQueue(final List<ResultSet> resultSets) throws SQLException {    // 遍历resultSets--在多少个目标实际表上执行SQL,该集合的size就有多大    for (ResultSet each : resultSets) {        // 将ResultSet和排序列信息封装成一个OrderByValue类型        OrderByValue orderByValue = new OrderByValue(each, orderByItems);        // 如果值存在,那么压入队列中        if (orderByValue.next()) {            orderByValuesQueue.offer(orderByValue);        }    }    // 重置currentResultSet的位置:如果队列不为空,那么将队列的顶部(peek)位置设置为currentResultSet的位置    setCurrentResultSet(orderByValuesQueue.isEmpty() ? resultSets.get(0) : orderByValuesQueue.peek().getResultSet());}@Overridepublic boolean next() throws SQLException {    // 调用next()判断是否还有值, 如果队列为空, 表示没有任何值, 那么直接返回false    if (orderByValuesQueue.isEmpty()) {        return false;    }    // 如果队列不为空, 那么第一次一定返回true;即有结果可取(且将isFirstNext置为false,表示接下来的请求都不是第一次请求next()方法)    if (isFirstNext) {        isFirstNext = false;        return true;    }    // 从队列中弹出第一个元素(因为是优先级队列,所以poll()返回的值,就是此次要取的值)    OrderByValue firstOrderByValue = orderByValuesQueue.poll();    // 如果它的next()存在,那么将它的next()再添加到队列中    if (firstOrderByValue.next()) {        orderByValuesQueue.offer(firstOrderByValue);    }    // 队列中所有元素全部处理完后就返回false    if (orderByValuesQueue.isEmpty()) {        return false;    }    // 再次重置currentResultSet的位置为队列的顶部位置;    setCurrentResultSet(orderByValuesQueue.peek().getResultSet());    return true;}

继续深入剖析:这段代码初看可能有点绕,假设运行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核心源码如下:

public LimitDecoratorResultSetMerger(final ResultSetMerger resultSetMerger, final Limit limit) throws SQLException {    super(resultSetMerger);    // limit赋值(Limit对象包括limit m,n中的m和n两个值)    this.limit = limit;    // 判断是否会跳过所有的结果项,即判断是否有符合条件的结果    skipAll = skipOffset();}private boolean skipOffset() throws SQLException {    // 假定limit.getOffsetValue()就是offset,实例sql中为limit 2,3,所以offset=2    for (int i = 0; i < limit.getOffsetValue(); i++) {        // 尝试从OrderByStreamResultSetMerger生成的优先级队列中跳过offset个元素,如果.next()一直为true,表示有足够符合条件的结果,那么返回false;否则没有足够符合条件的结果,那么返回true;即skilAll=true就表示跳过了所有没有符合条件的结果;        if (!getResultSetMerger().next()) {            return true;        }    }    // limit m,n的sql会被重写为limit 0, m+n,所以limit.isRowCountRewriteFlag()为true,rowNumber的值为0;    rowNumber = limit.isRowCountRewriteFlag() ? 0 : limit.getOffsetValue();    return false;}@Overridepublic boolean next() throws SQLException {    // 如果skipAll为true,即跳过所有,表示没有任何符合条件的值,那么返回false    if (skipAll) {        return false;    }    if (limit.getRowCountValue() > -1) {        // 每次next()获取值后,rowNumber自增,当自增rowCountValue次后,就不能再往下继续取值了,因为条件limit 2,3(rowCountValue=3)限制了        return ++rowNumber <= limit.getRowCountValue() && getResultSetMerger().next();    }    return getResultSetMerger().next();}

IteratorStreamResultSetMerger

构造方法核心源码:

private final Iterator<ResultSet> resultSets;public IteratorStreamResultSetMerger(final List<ResultSet> resultSets) {    // 将List<ResultSet>改成Iterator<ResultSet>,方便接下来迭代取得结果;    this.resultSets = resultSets.iterator();    // 重置currentResultSet    setCurrentResultSet(this.resultSets.next());}

END

原文发布于微信公众号 - Java技术驿站(chenssy89)

原文发表时间:2018-05-23

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏java思维导图

Java 单例以及单例所引发的思考

1 前言 前几天无意中看到一篇文章,讲到了老生常谈的单例,抱着复习一下的心态点了进去,还是那些熟悉的内容,可是却发现自己思考的角度变了,以前更多的是去记忆,只停...

2917
来自专栏Golang语言社区

go语言的sql包原理与用法分析

go的sql包是在pkg/database中,里面的两个包sql和sql/driver可以一起看。建议看这个两个包之前可以先看看sql文件夹下的doc.txt。...

3014
来自专栏逸鹏说道

如何让所有实体类用相同名称的主键(很有力的问题,比如所有表实体主键都用ID)

这个问题比较有力量,哈哈! 例如:有两个表userbases和products 两个表的主键分别为UserID和ProductID,那么,我想问有没有一种方法把...

3135
来自专栏我是攻城师

ElasticSearch+Solr几个case笔记

3104
来自专栏IMWeb前端团队

bash 的条件和循环

本文作者:IMWeb 江源 原文出处:IMWeb社区 未经同意,禁止转载 原文 条件语句和循环可以统称为流程控制,是一门语言最基础的部分。 bash ...

1896
来自专栏PHP在线

mysql学习笔记

InnoDB引擎与MyISAM引擎 mysql是关系型数据库。其中的存储引擎可以show engines来查看。我的版本是5.6.26的,查看版本用selec...

2855
来自专栏技术博客

Entity Framework DataAnnotations

DataAnnotation 特性由.NET 3.5中引进,给.NET中的类提供了一种添加验证的方式。但是在EF中它又可以对映射关系进行控制,相比较Fluent...

933
来自专栏吴伟祥

字段规范 原

541
来自专栏java学习

Java每日一练(2017/8/2)

本期题目: (单选题)1、在 java 中,一个类可同时定义为许多同名的方法,这些方法的形式参数个数,类型或顺序各不相同,传回的值可能个不相同,这种面向对象的...

2877
来自专栏Hongten

python开发_pickle

pickle模块使用的数据格式是python专用的,并且不同版本不向后兼容,同时也不能被其他语言说识别。要和其他语言交互,可以使用内置的json包使用pickl...

1022

扫码关注云+社区

领取腾讯云代金券