Sharding-JDBC 源码之 SQL 执行

男娘i 2022-12-31 02:29 128阅读 0赞

### Sharding-JDBC 系列 ###

1.  第一篇 [Sharding-JDBC 源码之启动流程分析][Sharding-JDBC]
2.  第二篇 [Sharding-JDBC 源码之 SQL 解析][Sharding-JDBC _ SQL]
3.  第三篇 [Sharding-JDBC 源码之 SQL 路由][Sharding-JDBC _ SQL 1]
4.  第四篇 [Sharding-JDBC 源码之 SQL 改写][Sharding-JDBC _ SQL 2]
5.  第五篇 [Sharding-JDBC 源码之 SQL 执行][Sharding-JDBC _ SQL 3]（本文）

--------------------

> Sharding-JDBC 使用执行引擎将路由和改写完成之后的真实SQL安全且高效发送到底层数据源执行。 它不是简单地将SQL通过JDBC直接发送至数据源执行；也并非直接将执行请求放入线程池去并发执行。它更关注平衡数据源连接创建以及内存占用所产生的消耗，以及最大限度地合理利用并发等问题。 执行引擎的目标是自动化的平衡资源控制与执行效率。

### 目录 ###

*   *   *  Sharding-JDBC 系列
         *  连接模式
         *   *  内存限制模式
             *  连接限制模式
         *  源码分析
         *   *  将路由结果转换为 PreparedStatementUnit 列表
             *  PreparedStatementExecutor 执行器 execute 方法
             *  ExecutorEngine 执行 execute 方法
             *  执行引擎中的 executeInternal
         *  总结

### 连接模式 ###

*  从资源控制的角度看，业务方访问数据库的连接数量应当有所限制。 防止某一个业务将数据库连接的资源耗尽，以致于影响其他业务的正常访问。 特别是一个不带分片键的逻辑SQL将落入到同库的多张真实表上。
 *  从执行效率的角度看，为每个分片查询维持一个独立的数据库连接，可以更加有效的利用多线程来提升执行效率。 为每个数据库连接开启独立的线程，可以将I/O所产生的消耗并行处理。为每个分片维持一个独立的数据库连接，还能够避免过早的将查询结果数据加载至内存。 独立的数据库连接，能够持有查询结果集游标位置的引用，在需要获取相应数据时移动游标即可。

#### 内存限制模式 ####

*  使用此模式的前提是，Sharding-JDBC 对一次操作所耗费的数据库连接数量不做限制；
 *  如果实际执行的SQL需要对某数据库实例中的200张表做操作，则对每张表创建一个新的数据库连接，并通过多线程的方式并发处理，以达成执行效率最大化。
 *  在SQL满足条件情况下，优先选择**流式归并**，以防止出现内存溢出或避免频繁垃圾回收情况。

#### 连接限制模式 ####

*  使用此模式的前提是，Sharding-JDBC 严格控制对一次操作所耗费的数据库连接数量。
 *  如果实际执行的SQL需要对某数据库实例中的200张表做操作，那么只会创建唯一的数据库连接，并对其200张表串行处理。
 *  如果一次操作中的分片散落在不同的数据库，仍然采用多线程处理对不同库的操作，但每个库的每次操作仍然只创建一个唯一的数据库连接。
 *  这样即可以防止对一次请求对数据库连接占用过多所带来的问题。该模式始终选择**内存归并**。

**内存限制模式**适用于OLAP操作，可以通过放宽对数据库连接的限制提升系统吞吐量；  
**连接限制模式**适用于OLTP操作，OLTP通常带有分片键，会路由到单一的分片，因此严格控制数据库连接，以保证在线系统数据库资源能够被更多的应用所使用，是明智的选择。

内存归并和流式归并的区别将在下一篇章介绍，敬请期待。

### 源码分析 ###

#### 将路由结果转换为 PreparedStatementUnit 列表 ####

public boolean execute() throws SQLException { 
            try { 
                // 获取路由映射 
                Collection<PreparedStatementUnit> preparedStatementUnits = route();
                // 2. 初始化 PreparedStatementExecutor 对象，并执行查询
                return new PreparedStatementExecutor(
                        getConnection().getShardingContext().getExecutorEngine(), routeResult.getSqlStatement().getType(), preparedStatementUnits).execute();
            } finally { 
                JDBCShardingRefreshHandler.build(routeResult, connection).execute();
                clearBatch();
            }
        }
    
        private Collection<PreparedStatementUnit> route() throws SQLException { 
            Collection<PreparedStatementUnit> result = new LinkedList<>();
            routeResult = routingEngine.route(getParameters());
            for (SQLExecutionUnit each : routeResult.getExecutionUnits()) { 
                PreparedStatement preparedStatement = generatePreparedStatement(each);
                routedStatements.add(preparedStatement);
                replaySetParameter(preparedStatement, each.getSqlUnit().getParameterSets().get(0));
                // 将 each 执行单元与 PreparedStatement映射封装到 result 对象中
                result.add(new PreparedStatementUnit(each, preparedStatement));
            }
            return result;
        }

1.  将路由改写后的执行单元与生成的 `PreparedStatement` 对象映射封装为 `PreparedStatementUnit` 对象；
2.  初始化 `PreparedStatementExecutor` 对象，传入**执行引擎**，**SQL 类型**，**执行单元**，并执行 `execute` 方法；

#### PreparedStatementExecutor 执行器 execute 方法 ####

public boolean execute() throws SQLException { 
            // 1. 实际由 executorEngine 执行
            List<Boolean> result = executorEngine.execute(sqlType, preparedStatementUnits, new ExecuteCallback<Boolean>() { 
                // 2. 注入回调函数
                @Override
                public Boolean execute(final BaseStatementUnit baseStatementUnit) throws Exception { 
                    return ((PreparedStatement) baseStatementUnit.getStatement()).execute();
                }
            });
            if (null == result || result.isEmpty() || null == result.get(0)) { 
                return false;
            }
            return result.get(0);
        }

1.  `PreparedStatementExecutor` 执行器 `execute` 方法的内部可以看到，实际是 `executorEngine` 引擎真正执行的；
2.  执行引擎在执行前，注入了一个回调函数 `ExecuteCallback`；

#### ExecutorEngine 执行 execute 方法 ####

public <T> List<T> execute(
                final SQLType sqlType, final Collection<? extends BaseStatementUnit> baseStatementUnits, final ExecuteCallback<T> executeCallback) throws SQLException { 
            // 如果没有可执行单元列表，则直接返回空
            if (baseStatementUnits.isEmpty()) { 
                return Collections.emptyList();
            }
            
            // 发送执行事件，供其他模块订阅
            OverallExecutionEvent event = new OverallExecutionEvent(sqlType, baseStatementUnits.size());
            EventBusInstance.getInstance().post(event);
            
            Iterator<? extends BaseStatementUnit> iterator =
            baseStatementUnits.iterator();
            // 1. 活动执行单元列表中的第一个
            BaseStatementUnit firstInput = iterator.next();
            // 2. 如果迭代器中还有其他元素则异步执行
            ListenableFuture<List<T>> restFutures = asyncExecute(sqlType, Lists.newArrayList(iterator), executeCallback);
            T firstOutput;
            List<T> restOutputs;
            try { 
                // 3. 同步执行第一个执行单元
                firstOutput = syncExecute(sqlType, firstInput, executeCallback);
                // 获取异步执行的结果
                restOutputs = restFutures.get();
            } catch (final Exception ex) { 
                event.setException(ex);
                event.setEventExecutionType(EventExecutionType.EXECUTE_FAILURE);
                EventBusInstance.getInstance().post(event);
                ExecutorExceptionHandler.handleException(ex);
                return null;
            }
            event.setEventExecutionType(EventExecutionType.EXECUTE_SUCCESS);
            EventBusInstance.getInstance().post(event);
            // 4. 组装执行结果
            List<T> result = Lists.newLinkedList(restOutputs);
            result.add(0, firstOutput);
            return result;
        }

1.  首先获取执行单元列表迭代器 `baseStatementUnits.iterator()`，并取出迭代器中第一个执行单元；
2.  将迭代器中剩下的元素重新生成一个列表，执行 `asyncExecute` 异步执行方法，该方法会遍历新生成列表中的元素，异步执行 `executeInternal`，并将结果装入 `restOutputs` 列表中；
3.  同步执行第一个执行单元；
4.  将第一个执行结果与异步执行的结果装入新的结果列表中；
5.  无论是同步执行还是异步执行，底层调用的方法都是一样的，都是 `executeInternal` 方法。

#### 执行引擎中的 executeInternal ####

private <T> T executeInternal(final SQLType sqlType, final BaseStatementUnit baseStatementUnit, final ExecuteCallback<T> executeCallback,
                                      final boolean isExceptionThrown, final Map<String, Object> dataMap) throws Exception { 
            // 1. 对当前连接对象上锁 
            synchronized (baseStatementUnit.getStatement().getConnection()) { 
                T result;
                ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
                ExecutorDataMap.setDataMap(dataMap);
                List<AbstractExecutionEvent> events = new LinkedList<>();
                for (List<Object> each : baseStatementUnit.getSqlExecutionUnit().getSqlUnit().getParameterSets()) { 
                    events.add(getExecutionEvent(sqlType, baseStatementUnit, each));
                }
                // 发送执行事件
                for (AbstractExecutionEvent event : events) { 
                    EventBusInstance.getInstance().post(event);
                }
                try { 
                    // 2. 执行回调方法
                    result = executeCallback.execute(baseStatementUnit);
                } catch (final SQLException ex) { 
                    for (AbstractExecutionEvent each : events) { 
                        each.setEventExecutionType(EventExecutionType.EXECUTE_FAILURE);
                        each.setException(ex);
                        EventBusInstance.getInstance().post(each);
                        ExecutorExceptionHandler.handleException(ex);
                    }
                    return null;
                }
                for (AbstractExecutionEvent each : events) { 
                    each.setEventExecutionType(EventExecutionType.EXECUTE_SUCCESS);
                    EventBusInstance.getInstance().post(each);
                }
                return result;
            }
        }

1.  对当前执行单元的连接上锁，即同一个连接一次只能执行一个请求，其他请求需要排队等待处理；
2.  执行回调方法，也是在回调方法中才会去真正的执行改写后的SQL，即从执行单元中获取实际的 `PreparedStatement` 对象，并执行 `execute` 方法，返回执行是否成功。
3.  `executeInternal` 方法执行完成后，依次出栈，回到最初调用的方法处：  
    `org.apache.ibatis.executor.statement.PreparedStatementHandler#query`

public <E> List<E> query(Statement statement, ResultHandler resultHandler) throws SQLException { 
        PreparedStatement ps = (PreparedStatement) statement;
        // 执行完成
        ps.execute();
        // 处理结果集
        return resultSetHandler.handleResultSets(ps);
      }

结果集处理下一篇章进行分析。

### 总结 ###

1.  从资源控制和执行效率两个角度可以将连接模式分为两种：

*  内存限制模式：连接数不限制，采用**流式归并**结果集来**避免内存溢出**或频繁垃圾回收；
 *  连接限制模式：**一次请求**只会对**同一个数据库**创建一个连接，分片到多个库，仍然保证每个库的每一次请求只会创建唯一的数据库连接，**防止数据库连接被占用过多**。

1.  真实 SQL 主要是通过调用 `executorEngine.execute` 执行的，在调用 `execute` 方法时会传入一个回调函数，**该回调函数中会去获取 Statement 并执行具体的 SQL 语句**；
2.  如果执行单元列表长度为 1，则会同步去执行SQL，如果长度大于 1，则**第一个会同步执行**，**剩下的将会异步执行**，最后合并同步执行和异步执行的结果集；
3.  对连接上**同步锁**，同一个连接一次只允许执行一个请求。

[Sharding-JDBC]: https://blog.csdn.net/smile_from_2015/article/details/111463029
[Sharding-JDBC _ SQL]: https://blog.csdn.net/smile_from_2015/article/details/111657901
[Sharding-JDBC _ SQL 1]: https://blog.csdn.net/smile_from_2015/article/details/111770894
[Sharding-JDBC _ SQL 2]: https://blog.csdn.net/smile_from_2015/article/details/111800577
[Sharding-JDBC _ SQL 3]: https://blog.csdn.net/smile_from_2015/article/details/111828140