devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Sharding-JDBC源码解析与vivo的定制开发_shardingconditions

作者：我家自动化 | 2024-04-12 10:18:56

赞

踩

shardingconditions

作者：vivo IT 平台团队 - Xiong Huanxin

Sharding-JDBC是在JDBC层提供服务的数据库中间件，在分库分表场景具有广泛应用。本文对Sharding-JDBC的解析、路由、改写、执行、归并五大核心引擎进行了源码解析，并结合业务实践经验，总结了使用Sharding-JDBC的一些痛点问题并分享了对应的定制开发与改造方案。

本文源码基于Sharding-JDBC 4.1.1版本。

一、业务背景

随着业务并发请求和数据规模的不断扩大，单节点库表压力往往会成为系统的性能瓶颈。公司IT内部营销库存、交易订单、财经台账、考勤记录等多领域的业务场景的日增数据量巨大，存在着数据库节点压力过大、连接过多、查询速度变慢等情况，根据数据来源、时间、工号等信息来将没有联系的数据尽量均分到不同的库表中，从而在不影响业务需求的前提下，减轻数据库节点压力，提升查询效率和系统稳定性

图片

二、技术选型

我们对比了几款比较常见的支持分库分表和读写分离的中间件。

图片

Sharding-JDBC作为轻量化的增强版的JDBC框架，相较其他中间件性能更好，接入难度更低，其数据分片、读写分离功能也覆盖了我们的业务诉求，因此我们在业务中广泛使用了Sharding-JDBC。但在使用Sharding-JDBC的过程中，我们也发现了诸多问题，为了业务更便捷的使用Sharding-JDBC，我们对源码做了针对性的定制开发和组件封装来满足业务需求。

图片

三、源码解析

3.1 引言

Sharding-JDBC作为基于JDBC的数据库中间件，实现了JDBC的标准api，Sharding-JDBC与原生JDBC的执行对比流程如下图所示：

图片

相关执行流程的代码样例如下：

  • JDBC执行样例

  1. //获取数据库连接
  2. try (Connection conn = DriverManager.getConnection("mysqlUrl", "userName", "password")) {
  3.     String sql = "SELECT * FROM  t_user WHERE name = ?";
  4.     //预编译SQL
  5.     try (PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
  6.         //参数设置与执行
  7.         preparedStatement.setString(1, "vivo");
  8.         preparedStatement.execute(sql);
  9.         //获取结果集
  10.         try (ResultSet resultSet = preparedStatement.getResultSet()) {
  11.             while (resultSet.next()) {
  12.                 //处理结果
  13.             }
  14.         }
  15.     }
  16. }

  • Sharding-JDBC 源码

  1. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement#execute
  2.     public boolean execute() throws SQLException {
  3.         try {
  4.             clearPrevious();
  5.             //解析+路由+重写 内部调用BasePrepareEngine#prepare方法
  6.             prepare();
  7.             initPreparedStatementExecutor();
  8.             //执行
  9.             return preparedStatementExecutor.execute();
  10.         } finally {
  11.             clearBatch();
  12.         }
  13.     }
  14.  
  15. org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine#prepare
  16.     public ExecutionContext prepare(final String sql, final List<Object> parameters) {
  17.         List<Object> clonedParameters = cloneParameters(parameters);
  18.         //解析+路由(executeRoute内部先进行解析再执行路由)
  19.         RouteContext routeContext = executeRoute(sql, clonedParameters);
  20.         ExecutionContext result = new ExecutionContext(routeContext.getSqlStatementContext());
  21.         //重写
  22.         result.getExecutionUnits().addAll(executeRewrite(sql, clonedParameters, routeContext));
  23.         if (properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SHOW)) {
  24.             SQLLogger.logSQL(sql, properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SIMPLE), result.getSqlStatementContext(), result.getExecutionUnits());
  25.         }
  26.         return result;
  27.     }
  28.  
  29. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#getResultSet
  30.     public ResultSet getResultSet() throws SQLException {
  31.         if (null != currentResultSet) {
  32.             return currentResultSet;
  33.         }
  34.         if (executionContext.getSqlStatementContext() instanceof SelectStatementContext || executionContext.getSqlStatementContext().getSqlStatement() instanceof DALStatement) {
  35.             List<ResultSet> resultSets = getResultSets();
  36.             //归并结果集
  37.             MergedResult mergedResult = mergeQuery(getQueryResults(resultSets));
  38.             currentResultSet = new ShardingResultSet(resultSets, mergedResult, this, executionContext);
  39.         }
  40.         return currentResultSet;
  41.     }

从对比的执行流程图可见：

  • 【JDBC】：执行的主要流程是通过Datasource获取Connection，再注入SQL语句生成PreparedStatement对象，PreparedStatement设置占位符参数执行后得到结果集ResultSet。

  • 【Sharding-JDBC】：主要流程基本一致，但Sharding基于PreparedStatement进行了实现与扩展，具体实现类ShardingPreparedStatement中会抽象出解析、路由、重写、归并等引擎，从而实现分库分表、读写分离等能力，每个引擎的作用说明如下表所示：

图片

//*相关引擎的源码解析在下文会作更深入的阐述

3.2 解析引擎

3.2.1 引擎解析

解析引擎是Sharding-JDBC进行分库分表逻辑的基础，其作用是将SQL拆解为不可再分的原子符号（称为token），再根据数据库类型将这些token分类成关键字、表达式、操作符、字面量等不同类型，进而生成抽象语法树，而语法树是后续进行路由、改写操作的前提（这也正是语法树的存在使得Sharding-JDBC存在各式各样的语法限制的原因之一）。

图片

▲图片来源：ShardingSphere 官方文档

4.x的版本采用ANTLR（ANother Tool for Language Recognition）作为解析引擎，在ShardingSphere-sql-parser-dialect模块中定义了适用于不同数据库语法的解析规则（.g4文件），idea中也可以下载ANTLR v4的插件，输入SQL查看解析后的语法树结果。

图片

解析方法的入口在DataNodeRouter的createRouteContext方法中，解析引擎根据数据库类型和SQL创建SQLParserExecutor执行得到解析树，再通过ParseTreeVisitor()的visit方法，对解析树进行处理得到SQLStatement。ANTLR支持listener和visitor两种模式的接口，visitor方式可以更灵活的控制解析树的遍历过程，更适用于SQL解析的场景。

  • 解析引擎核心代码

  1. org.apache.shardingsphere.underlying.route.DataNodeRouter#createRouteContext#96
  2.     private RouteContext createRouteContext(final String sql, final List<Object> parameters, final boolean useCache) {
  3.         //解析引擎解析SQL
  4.         SQLStatement sqlStatement = parserEngine.parse(sql, useCache);
  5.         try {
  6.             SQLStatementContext sqlStatementContext = SQLStatementContextFactory.newInstance(metaData.getSchema(), sql, parameters, sqlStatement);
  7.             return new RouteContext(sqlStatementContext, parameters, new RouteResult());
  8.             // TODO should pass parameters for master-slave
  9.         } catch (final IndexOutOfBoundsException ex) {
  10.             return new RouteContext(new CommonSQLStatementContext(sqlStatement), parameters, new RouteResult());
  11.         }
  12.     }
  13.  
  14. org.apache.shardingsphere.sql.parser.SQLParserEngine#parse0#72
  15.     private SQLStatement parse0(final String sql, final boolean useCache) {
  16.         //缓存
  17.         if (useCache) {
  18.             Optional<SQLStatement> cachedSQLStatement = cache.getSQLStatement(sql);
  19.             if (cachedSQLStatement.isPresent()) {
  20.                 return cachedSQLStatement.get();
  21.             }
  22.         }
  23.         //根据数据库类型和sql生成解析树
  24.         ParseTree parseTree = new SQLParserExecutor(databaseTypeName, sql).execute().getRootNode();
  25.         //ParseTreeVisitor的visit方法对解析树进行处理得到SQLStatement
  26.       SQLStatement result = (SQLStatement) ParseTreeVisitorFactory.newInstance(databaseTypeName, VisitorRule.valueOf(parseTree.getClass())).visit(parseTree);
  27.         if (useCache) {
  28.             cache.put(sql, result);
  29.         }
  30.         return result;
  31.     }

SQLStatement实际上是一个接口，其实现对应着不同的SQL类型，如SelectStatement 类中就包括查询的字段、表名、where条件、分组、排序、分页、lock等变量，可以看到这里并没有对having这种字段做定义，相当于Sharding-JDBC无法识别到SQL中的having，这使得Sharding-JDBC对having语法有一定的限制。

  • SelectStatement

  1. public final class SelectStatement extends DMLStatement {
  2.     // 字段
  3.     private ProjectionsSegment projections;
  4.     // 表
  5.     private final Collection<TableReferenceSegment> tableReferences = new LinkedList<>();
  6.     // where
  7.     private WhereSegment where;
  8.     // groupBy
  9.     private GroupBySegment groupBy;
  10.     // orderBy
  11.     private OrderBySegment orderBy;
  12.     // limit
  13.     private LimitSegment limit;
  14.     // 父statement
  15.     private SelectStatement parentStatement;
  16.     // lock
  17.     private LockSegment lock;
  18. }

SQLStatement还会被进一步转换成SQLStatementContext，如SelectStatement 会被转换成SelectStatementContext ，其结构与SelectStatement 类似不再多说，值得注意的是虽然这里定义了containsSubquery来判断是否包含子查询，但4.1.1源码永远是返回的false，与having类似，这意味着Sharding-JDBC不会对子查询语句做特殊处理。

  • SelectStatementContext

  1. public final class SelectStatementContext extends CommonSQLStatementContext<SelectStatement> implements TableAvailable, WhereAvailable {
  2.      
  3.     private final TablesContext tablesContext;
  4.      
  5.     private final ProjectionsContext projectionsContext;
  6.      
  7.     private final GroupByContext groupByContext;
  8.      
  9.     private final OrderByContext orderByContext;
  10.      
  11.     private final PaginationContext paginationContext;
  12.      
  13.     private final boolean containsSubquery;
  14. }
  15.  
  16.     private boolean containsSubquery() {
  17.         // FIXME process subquery
  18. //        Collection<SubqueryPredicateSegment> subqueryPredicateSegments = getSqlStatement().findSQLSegments(SubqueryPredicateSegment.class);
  19. //        for (SubqueryPredicateSegment each : subqueryPredicateSegments) {
  20. //            if (!each.getAndPredicates().isEmpty()) {
  21. //                return true;
  22. //            }
  23. //        }
  24.         return false;
  25.     }

3.2.2 引擎总结

解析引擎是进行路由改写的前提基础，其作用就是将SQL按照定义的语法规则拆分成原子符号(token)，生成语法树，根据不同的SQL类型生成对应的SQLStatement，SQLStatement由各自的Segment组成，所有的Segment都包含startIndex和endIndex来定位token在SQL中所属的位置，但解析语法难以涵盖所有的SQL场景，使得部分SQL无法按照预期的结果路由执行。

3.3 路由引擎

3.3.1 引擎解析

路由引擎是Sharding-JDBC的核心步骤，作用是根据定义的分库分表规则将解析引擎生成的SQL上下文生成对应的路由结果，RouteResult 包括DataNode和RouteUnit，DataNode是实际的数据源节点，包括数据源名称和实际的物理表名，RouteUnit则记录了逻辑表/库与物理表/库的映射关系，后面的改写引擎也是根据这个映射关系来决定如何替换SQL中的逻辑表（实际上RouteResult 就是维护了一条SQL需要往哪些库哪些表执行的关系）。

  • RouteResult

  1. public final class RouteResult {
  2.      
  3.     private final Collection<Collection<DataNode>> originalDataNodes = new LinkedList<>();
  4.      
  5.     private final Collection<RouteUnit> routeUnits = new LinkedHashSet<>();
  6. }
  7.  
  8. public final class DataNode {
  9.      
  10.     private static final String DELIMITER = ".";
  11.      
  12.     private final String dataSourceName;
  13.      
  14.     private final String tableName;
  15. }
  16.  
  17. public final class RouteUnit {
  18.      
  19.     private final RouteMapper dataSourceMapper;
  20.      
  21.     private final Collection<RouteMapper> tableMappers;
  22. }
  23.  
  24. public final class RouteMapper {
  25.      
  26.     private final String logicName;
  27.      
  28.     private final String actualName;
  29. }

其中，路由有分为分片路由和主从路由，两者可以单独使用，也可以组合使用。

  • 分片路由

ShardingRouteDecorator的decorate方法是路由引擎的核心逻辑，经过SQL校验->生成分片条件->合并分片值后得到路由结果。

  • 分片路由decorate方法

  1. org.apache.shardingsphere.sharding.route.engine.ShardingRouteDecorator#decorate#57
  2. public RouteContext decorate(final RouteContext routeContext, final ShardingSphereMetaData metaData, final ShardingRule shardingRule, final ConfigurationProperties properties) {
  3.         SQLStatementContext sqlStatementContext = routeContext.getSqlStatementContext();
  4.         List<Object> parameters = routeContext.getParameters();
  5.         //SQL校验  校验INSERT INTO .... ON DUPLICATE KEY UPDATE 和UPDATE语句中是否存在分片键
  6.       ShardingStatementValidatorFactory.newInstance(
  7.                 sqlStatementContext.getSqlStatement()).ifPresent(validator -> validator.validate(shardingRule, sqlStatementContext.getSqlStatement(), parameters));
  8.         //生成分片条件
  9.         ShardingConditions shardingConditions = getShardingConditions(parameters, sqlStatementContext, metaData.getSchema(), shardingRule);
  10.         //合并分片值
  11.         boolean needMergeShardingValues = isNeedMergeShardingValues(sqlStatementContext, shardingRule);
  12.         if (sqlStatementContext.getSqlStatement() instanceof DMLStatement && needMergeShardingValues) {
  13.             checkSubqueryShardingValues(sqlStatementContext, shardingRule, shardingConditions);
  14.             mergeShardingConditions(shardingConditions);
  15.         }
  16.         ShardingRouteEngine shardingRouteEngine = ShardingRouteEngineFactory.newInstance(shardingRule, metaData, sqlStatementContext, shardingConditions, properties);
  17.         //得到路由结果
  18.         RouteResult routeResult = shardingRouteEngine.route(shardingRule);
  19.         if (needMergeShardingValues) {
  20.             Preconditions.checkState(1 == routeResult.getRouteUnits().size(), "Must have one sharding with subquery.");
  21.         }
  22.         return new RouteContext(sqlStatementContext, parameters, routeResult);
  23.     }

ShardingStatementValidator有ShardingInsertStatementValidator和ShardingUpdateStatementValidator两种实现，INSERT INTO .... ON DUPLICATE KEY UPDATE和UPDATE语法都会涉及到字段值的更新，Sharding-JDBC是不允许更新分片值的，毕竟修改分片值还需要将数据迁移至新分片值对应的库表中，才能保证数据分片规则一致。两者的校验细节也有所不同：

  • INSERT INTO .... ON DUPLICATE KEY UPDATE仅仅是对UPDATE字段的校验， ON DUPLICATE KEY UPDATE中包含分片键就会报错；

  • 而UPDATE语句则会额外校验WHERE条件中分片键的原始值和SET的值是否一样，不一样则会抛出异常。

图片

ShardingCondition中只有一个变量routeValues，RouteValue是一个接口，有ListRouteValue和RangeRouteValue两种实现，前者记录了分片键的in或=条件的分片值，后者则记录了范围查询的分片值，两者被封装为ShardingValue对象后，将会透传至分片算法中计算得到分片结果集。

  • ShardingCondition

  1. public final class ShardingConditions {
  2.      
  3.     private final List<ShardingCondition> conditions;
  4. }
  5.  
  6. public class ShardingCondition {
  7.      
  8.     private final List<RouteValue> routeValues = new LinkedList<>();
  9. }
  10.  
  11.  
  12. public final class ListRouteValue<T extends Comparable<?>> implements RouteValue {
  13.      
  14.     private final String columnName;
  15.      
  16.     private final String tableName;
  17.     //in或=条件对应的值
  18.     private final Collection<T> values;
  19.      
  20.     @Override
  21.     public String toString() {
  22.         return tableName + "." + columnName + (1 == values.size() ? " = " + new ArrayList<>(values).get(0) : " in (" + Joiner.on(",").join(values) + ")");
  23.     }
  24. }
  25.  
  26. public final class RangeRouteValue<T extends Comparable<?>> implements RouteValue {
  27.      
  28.     private final String columnName;
  29.      
  30.     private final String tableName;
  31.     //between and 大于小于等范围值的上下限
  32.     private final Range<T> valueRange;
  33. }

生成分片条件后还会合并分片条件，但是前文提过在SelectStatementContext中的containsSubquery永远是false，所以这段逻辑永远返回false，即不会合并分片条件。

  • 判断是否需要合并分片条件

  1. org.apache.shardingsphere.sharding.route.engine.ShardingRouteDecorator#isNeedMergeShardingValues#87
  2. private boolean isNeedMergeShardingValues(final SQLStatementContext sqlStatementContext, final ShardingRule shardingRule) {
  3.         return sqlStatementContext instanceof SelectStatementContext && ((SelectStatementContext) sqlStatementContext).isContainsSubquery()
  4.                 && !shardingRule.getShardingLogicTableNames(sqlStatementContext.getTablesContext().getTableNames()).isEmpty();
  5.     }

然后就是通过分片路由引擎调用分片算法计算路由结果了，ShardingRouteEngine实现较多，介绍起来篇幅较多，这里就不展开说明了，可以参考官方文档来了解路由引擎的选择规则。

图片

▲图片来源：ShardingSphere 官方文档

Sharding-JDBC定义了多种分片策略和算法接口，主要的分配策略与算法说明如下表所示：

图片

补充两个细节：

（1）当ALLOW_RANGE_QUERY_WITH_INLINE_SHARDING配置设置true时，InlineShardingStrategy支持范围查询，但是并不是根据分片值计算范围，而是直接全路由至配置的数据节点，会存在性能隐患。

  • InlineShardingStrategy.doSharding

  1. org.apache.shardingsphere.core.strategy.route.inline.InlineShardingStrategy#doSharding
  2.     public Collection<String> doSharding(final Collection<String> availableTargetNames, final Collection<RouteValue> shardingValues, final ConfigurationProperties properties) {
  3.         RouteValue shardingValue = shardingValues.iterator().next();
  4.         //ALLOW_RANGE_QUERY_WITH_INLINE_SHARDING设置为true，直接返回availableTargetNames，而不是根据RangeRouteValue计算
  5.         if (properties.<Boolean>getValue(ConfigurationPropertyKey.ALLOW_RANGE_QUERY_WITH_INLINE_SHARDING) && shardingValue instanceof RangeRouteValue) {
  6.             return availableTargetNames;
  7.         }
  8.         Preconditions.checkState(shardingValue instanceof ListRouteValue, "Inline strategy cannot support this type sharding:" + shardingValue.toString());
  9.         Collection<String> shardingResult = doSharding((ListRouteValue) shardingValue);
  10.         Collection<String> result = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
  11.         for (String each : shardingResult) {
  12.             if (availableTargetNames.contains(each)) {
  13.                 result.add(each);
  14.             }
  15.         }
  16.         return result;
  17.     }

（2）4.1.1的官方文档虽然说Hint可以跳过解析和改写，但在我们上面解析引擎的源码解析中，我们并没有看到有对Hint策略的额外跳过。事实上，即使使用了Hint分片SQL也同样需要解析重写，也同样受Sharding-JDBC的语法限制，这在官方的issue中也曾经被提及。

图片

▲图片来源：ShardingSphere 官方文档

  • 主从路由

主从路由的核心逻辑就是通过MasterSlaveDataSourceRouter的route方法进行判定SQL走主库还是从库。主从情况下，配置的数据源实际是一组主从，而不是单个的实例，所以需要通过masterSlaveRule获取到具体的主库或者从库名字。

  • 主从路由decorate

  1. org.apache.shardingsphere.masterslave.route.engine.MasterSlaveRouteDecorator#decorate    
  2.     public RouteContext decorate(final RouteContext routeContext, final ShardingSphereMetaData metaData, final MasterSlaveRule masterSlaveRule, final ConfigurationProperties properties) {
  3.         //为空证明没有经过分片路由
  4.         if (routeContext.getRouteResult().getRouteUnits().isEmpty()) {
  5.             //根据SQL判断选择走主库还是从库
  6.             String dataSourceName = new MasterSlaveDataSourceRouter(masterSlaveRule).route(routeContext.getSqlStatementContext().getSqlStatement());
  7.             RouteResult routeResult = new RouteResult();
  8.            //根据具体的主库/从库名创建路由单元
  9.             routeResult.getRouteUnits().add(new RouteUnit(new RouteMapper(dataSourceName, dataSourceName), Collections.emptyList()));
  10.             return new RouteContext(routeContext.getSqlStatementContext(), Collections.emptyList(), routeResult);
  11.         }
  12.         Collection<RouteUnit> toBeRemoved = new LinkedList<>();
  13.         Collection<RouteUnit> toBeAdded = new LinkedList<>();
  14.         //不为空证明已经被分片路由处理了
  15.         for (RouteUnit each : routeContext.getRouteResult().getRouteUnits()) {
  16.             if (masterSlaveRule.getName().equalsIgnoreCase(each.getDataSourceMapper().getActualName())) {
  17.                 //先标记移除 因为这里是一组主从的名字而不是实际的库
  18.                 toBeRemoved.add(each);
  19.                 //根据SQL判断选择走主库还是从库
  20.                 String actualDataSourceName = new MasterSlaveDataSourceRouter(masterSlaveRule).route(routeContext.getSqlStatementContext().getSqlStatement());
  21.                 //根据具体的主库/从库名创建路由单元
  22.                 toBeAdded.add(new RouteUnit(new RouteMapper(each.getDataSourceMapper().getLogicName(), actualDataSourceName), each.getTableMappers()));
  23.             }
  24.         }
  25.         routeContext.getRouteResult().getRouteUnits().removeAll(toBeRemoved);
  26.         routeContext.getRouteResult().getRouteUnits().addAll(toBeAdded);
  27.         return routeContext;
  28.     }

MasterSlaveDataSourceRouter中isMasterRoute方法会判断SQL是否需要走主库，当出现以下情况时走主库：

  • select语句包含锁，如for update语句

  • 不是select语句

  • MasterVisitedManager.isMasterVisited()设置为true

  • HintManager.isMasterRouteOnly()设置为true

不走主库则通过负载算法选择从库，Sharding-JDBC提供了轮询和随机两种算法。

  • MasterSlaveDataSourceRouter

  1. public final class MasterSlaveDataSourceRouter {
  2.      
  3.     private final MasterSlaveRule masterSlaveRule;
  4.      
  5.     /**
  6.      * Route.
  7.      *
  8.      * @param sqlStatement SQL statement
  9.      * @return data source name
  10.      */
  11.     public String route(final SQLStatement sqlStatement) {
  12.         if (isMasterRoute(sqlStatement)) {
  13.             MasterVisitedManager.setMasterVisited();
  14.             return masterSlaveRule.getMasterDataSourceName();
  15.         }
  16.         return masterSlaveRule.getLoadBalanceAlgorithm().getDataSource(
  17.                 masterSlaveRule.getName(), masterSlaveRule.getMasterDataSourceName(), new ArrayList<>(masterSlaveRule.getSlaveDataSourceNames()));
  18.     }
  19.      
  20.     private boolean isMasterRoute(final SQLStatement sqlStatement) {
  21.         return containsLockSegment(sqlStatement) || !(sqlStatement instanceof SelectStatement) || MasterVisitedManager.isMasterVisited() || HintManager.isMasterRouteOnly();
  22.     }
  23.      
  24.     private boolean containsLockSegment(final SQLStatement sqlStatement) {
  25.         return sqlStatement instanceof SelectStatement && ((SelectStatement) sqlStatement).getLock().isPresent();
  26.     }
  27. }

是否走主库的信息存在MasterVisitedManager中，MasterVisitedManager是通过ThreadLocal实现的，但这种实现会有一个问题，当我们使用事务先查询再更新/插入时，第一条查询SQL并不会走主库，而是走从库，如果业务需要事务的第一条查询也走主库，事务查询前需要手动调用一次MasterVisitedManager.setMasterVisited()。

  • MasterVisitedManager

  1. public final class MasterVisitedManager {
  2.      
  3.     private static final ThreadLocal<Boolean> MASTER_VISITED = ThreadLocal.withInitial(() -> false);
  4.      
  5.     /**
  6.      * Judge master data source visited in current thread.
  7.      *
  8.      * @return master data source visited or not in current thread
  9.      */
  10.     public static boolean isMasterVisited() {
  11.         return MASTER_VISITED.get();
  12.     }
  13.      
  14.     /**
  15.      * Set master data source visited in current thread.
  16.      */
  17.     public static void setMasterVisited() {
  18.         MASTER_VISITED.set(true);
  19.     }
  20.      
  21.     /**
  22.      * Clear master data source visited.
  23.      */
  24.     public static void clear() {
  25.         MASTER_VISITED.remove();
  26.     }
  27. }

3.3.2 引擎总结

路由引擎的作用是将SQL根据参数通过实现的策略算法计算出实际该在哪些库的哪些表执行，也就是路由结果。路由引擎有两种实现，分别是分片路由和主从路由，两者都提供了标准化的策略接口来让业务实现自己的路由策略，分片路由需要注意自身SQL场景和策略算法相匹配，主从路由中同一线程且同一数据库连接内，有写入操作后，之后的读操作会从主库读取，写入操作前的读操作不会走主库。

3.4 改写引擎

3.4.1 引擎解析

经过解析路由后虽然确定了执行的实际库表，但SQL中表名依旧是逻辑表，不能执行，改写引擎可以将逻辑表替换为物理表。同时，路由至多库表的SQL也需要拆分为多条SQL执行。

改写的入口仍旧在BasePrepareEngine中，创建重写上下文createSQLRewriteContext，再根据上下文进行改写rewrite，最终返回执行单元ExecutionUnit。

  • 改写逻辑入口

  1. org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine#executeRewrite
  2.     private Collection<ExecutionUnit> executeRewrite(final String sql, final List<Object> parameters, final RouteContext routeContext) {
  3.         //注册重写装饰器
  4.         registerRewriteDecorator();
  5.         //创建 SQLRewriteContext
  6.         SQLRewriteContext sqlRewriteContext = rewriter.createSQLRewriteContext(sql, parameters, routeContext.getSqlStatementContext(), routeContext);
  7.         //重写
  8.         return routeContext.getRouteResult().getRouteUnits().isEmpty() ? rewrite(sqlRewriteContext) : rewrite(routeContext, sqlRewriteContext);
  9.     }

执行单元包含了数据源名称，改写后的SQL，以及对应的参数，SQL一样的两个SQLUnit会被视为相等。

  • ExecutionUnit

  1. @RequiredArgsConstructor
  2. @Getter
  3. @EqualsAndHashCode
  4. @ToString
  5. public final class ExecutionUnit {
  6.      
  7.     private final String dataSourceName;
  8.      
  9.     private final SQLUnit sqlUnit;
  10. }
  11.  
  12. @AllArgsConstructor
  13. @RequiredArgsConstructor
  14. @Getter
  15. @Setter
  16. //根据sql判断是否相等
  17. @EqualsAndHashCode(of = { "sql" })
  18. @ToString
  19. public final class SQLUnit {
  20.  
  21.     private String sql;
  22.  
  23.     private final List<Object> parameters;
  24.  
  25. }

createSQLRewriteContext完成了两件事，一个是对SQL参数进行了重写，一个是生成了SQLToken。

  • createSQLRewriteContext

  1. org.apache.shardingsphere.underlying.rewrite.SQLRewriteEntry#createSQLRewriteContext
  2.     public SQLRewriteContext createSQLRewriteContext(final String sql, final List<Object> parameters, final SQLStatementContext sqlStatementContext, final RouteContext routeContext) {
  3.         SQLRewriteContext result = new SQLRewriteContext(schemaMetaData, sqlStatementContext, sql, parameters);
  4.         //sql参数重写
  5.         decorate(decorators, result, routeContext);
  6.         //生成SQLToken
  7.         result.generateSQLTokens();
  8.         return result;
  9.     }
  10.  
  11. org.apache.shardingsphere.sharding.rewrite.context.ShardingSQLRewriteContextDecorator#decorate
  12.     public void decorate(final ShardingRule shardingRule, final ConfigurationProperties properties, final SQLRewriteContext sqlRewriteContext) {
  13.         for (ParameterRewriter each : new ShardingParameterRewriterBuilder(shardingRule, routeContext).getParameterRewriters(sqlRewriteContext.getSchemaMetaData())) {
  14.             if (!sqlRewriteContext.getParameters().isEmpty() && each.isNeedRewrite(sqlRewriteContext.getSqlStatementContext())) {
  15.                 //参数重写
  16.                 each.rewrite(sqlRewriteContext.getParameterBuilder(), sqlRewriteContext.getSqlStatementContext(), sqlRewriteContext.getParameters());
  17.             }
  18.         }
  19.         //sqlTokenGenerators
  20.         sqlRewriteContext.addSQLTokenGenerators(new ShardingTokenGenerateBuilder(shardingRule, routeContext).getSQLTokenGenerators());
  21.     }
  22.  
  23. org.apache.shardingsphere.underlying.rewrite.context.SQLRewriteContext#generateSQLTokens
  24.     public void generateSQLTokens() {
  25.         sqlTokens.addAll(sqlTokenGenerators.generateSQLTokens(sqlStatementContext, parameters, schemaMetaData));
  26.     }

ParameterRewriter中与分片相关的实现有两种。

图片

//*详细的例子可以参考官方文档中分页修正和补列部分。

SQLToken记录了SQL中每个token（解析引擎中提过的不可再分的原子符号）的起始位置，从而方便改写引擎知道哪些位置需要改写。

  • SQLToken

  1. @RequiredArgsConstructor
  2. @Getter
  3. public abstract class SQLToken implements Comparable<SQLToken> {
  4.      
  5.     private final int startIndex;
  6.      
  7.     @Override
  8.     public final int compareTo(final SQLToken sqlToken) {
  9.         return startIndex - sqlToken.getStartIndex();
  10.     }
  11. }

创建完SQLRewriteContext后就对整条SQL进行重写和组装参数，可以看出每个RouteUnit都会重写SQL并获取自己对应的参数。

  • SQLRouteRewriteEngine.rewrite

  1. org.apache.shardingsphere.underlying.rewrite.engine.SQLRouteRewriteEngine#rewrite
  2.     public Map<RouteUnit, SQLRewriteResult> rewrite(final SQLRewriteContext sqlRewriteContext, final RouteResult routeResult) {
  3.         Map<RouteUnit, SQLRewriteResult> result = new LinkedHashMap<>(routeResult.getRouteUnits().size(), 1);
  4.         for (RouteUnit each : routeResult.getRouteUnits()) {
  5.             //重写SQL+组装参数
  6.             result.put(each, new SQLRewriteResult(new RouteSQLBuilder(sqlRewriteContext, each).toSQL(), getParameters(sqlRewriteContext.getParameterBuilder(), routeResult, each)));
  7.         }
  8.         return result;
  9.     }

toSQL核心就是根据SQLToken将SQL拆分改写再拼装，比如select * from t_order where created_by = '123' 就会被拆分为select * from | t_order | where created_by = '123'三部分进行改写拼装。

  • toSQL

  1. org.apache.shardingsphere.underlying.rewrite.sql.impl.AbstractSQLBuilder#toSQL
  2.     public final String toSQL() {
  3.         if (context.getSqlTokens().isEmpty()) {
  4.             return context.getSql();
  5.         }
  6.         Collections.sort(context.getSqlTokens());
  7.         StringBuilder result = new StringBuilder();
  8.         //截取第一个SQLToken之前的内容  select * from
  9.         result.append(context.getSql().substring(0, context.getSqlTokens().get(0).getStartIndex()));
  10.         for (SQLToken each : context.getSqlTokens()) {
  11.             //重写拼接每个SQLToken对应的内容  t_order ->t_order_0
  12.             result.append(getSQLTokenText(each));
  13.             //拼接SQLToken中间不变的内容 where created_by = '123'
  14.             result.append(getConjunctionText(each));
  15.         }
  16.         return result.toString();
  17.     }

ParameterBuilder有StandardParameterBuilder和GroupedParameterBuilder两个实现。

  • StandardParameterBuilder：适用于非insert语句，getParameters无需分组处理直接返回即可

  • GroupedParameterBuilder：适用于insert语句，需要根据路由情况对参数进行分组。

原因和样例可以参考官方文档批量拆分部分。

  • getParameters

  1. org.apache.shardingsphere.underlying.rewrite.engine.SQLRouteRewriteEngine#getParameters
  2.     private List<Object> getParameters(final ParameterBuilder parameterBuilder, final RouteResult routeResult, final RouteUnit routeUnit) {
  3.         if (parameterBuilder instanceof StandardParameterBuilder || routeResult.getOriginalDataNodes().isEmpty() || parameterBuilder.getParameters().isEmpty()) {
  4.             //非插入语句直接返回
  5.             return parameterBuilder.getParameters();
  6.         }
  7.         List<Object> result = new LinkedList<>();
  8.         int count = 0;
  9.         for (Collection<DataNode> each : routeResult.getOriginalDataNodes()) {
  10.             if (isInSameDataNode(each, routeUnit)) {
  11.                 //插入语句参数分组构造
  12.                 result.addAll(((GroupedParameterBuilder) parameterBuilder).getParameters(count));
  13.             }
  14.             count++;
  15.         }
  16.         return result;
  17.     }

3.4.2 引擎总结

改写引擎的作用是将逻辑SQL转换为实际可执行的SQL，这其中既有逻辑表名的替换，也有多路由的SQL拆分，还有为了后续归并操作而进行的分页、分组、排序等改写，select语句不会对参数进行重组，而insert语句为了避免插入多余数据，会通过路由单元对参数进行重组。

3.5 执行引擎

3.5.1 引擎解析

改写完成后的SQL就可以执行了，执行引擎需要平衡好资源和效率，如果为每条真实SQL都创建一个数据库连接显然会造成资源的滥用，但如果单线程串行也必然会影响执行效率。

执行引擎会先将执行单元中需要执行的SQLUnit根据数据源分组，同一个数据源下的SQLUnit会放入一个list，然后会根据maxConnectionsSizePerQuery对同一个数据源的SQLUnit继续分组，创建连接并绑定SQLUnit 。

  • 执行组创建

  1. org.apache.shardingsphere.sharding.execute.sql.prepare.SQLExecutePrepareTemplate#getSynchronizedExecuteUnitGroups
  2.     private Collection<InputGroup<StatementExecuteUnit>> getSynchronizedExecuteUnitGroups(
  3.             final Collection<ExecutionUnit> executionUnits, final SQLExecutePrepareCallback callback) throws SQLException {
  4.         //根据数据源将SQLUnit分组 key=dataSourceName
  5.         Map<String, List<SQLUnit>> sqlUnitGroups = getSQLUnitGroups(executionUnits);
  6.         Collection<InputGroup<StatementExecuteUnit>> result = new LinkedList<>();
  7.         //创建sql执行组
  8.         for (Entry<String, List<SQLUnit>> entry : sqlUnitGroups.entrySet()) {
  9.             result.addAll(getSQLExecuteGroups(entry.getKey(), entry.getValue(), callback));
  10.         }
  11.         return result;
  12.     }
  13.  
  14. org.apache.shardingsphere.sharding.execute.sql.prepare.SQLExecutePrepareTemplate#getSQLExecuteGroups
  15.     private List<InputGroup<StatementExecuteUnit>> getSQLExecuteGroups(final String dataSourceName,
  16.                                                                        final List<SQLUnit> sqlUnits, final SQLExecutePrepareCallback callback) throws SQLException {
  17.         List<InputGroup<StatementExecuteUnit>> result = new LinkedList<>();
  18.         //每个连接需要执行的最大sql数量
  19.         int desiredPartitionSize = Math.max(0 == sqlUnits.size() % maxConnectionsSizePerQuery ? sqlUnits.size() / maxConnectionsSizePerQuery : sqlUnits.size() / maxConnectionsSizePerQuery + 1, 1);
  20.         //分组，每组对应一条数据库连接
  21.         List<List<SQLUnit>> sqlUnitPartitions = Lists.partition(sqlUnits, desiredPartitionSize);
  22.         //选择连接模式 连接限制/内存限制
  23.         ConnectionMode connectionMode = maxConnectionsSizePerQuery < sqlUnits.size() ? ConnectionMode.CONNECTION_STRICTLY : ConnectionMode.MEMORY_STRICTLY;
  24.         //创建连接
  25.         List<Connection> connections = callback.getConnections(connectionMode, dataSourceName, sqlUnitPartitions.size());
  26.         int count = 0;
  27.         for (List<SQLUnit> each : sqlUnitPartitions) {
  28.             //绑定连接和SQLUnit 创建StatementExecuteUnit
  29.             result.add(getSQLExecuteGroup(connectionMode, connections.get(count++), dataSourceName, each, callback));
  30.         }
  31.         return result;
  32.     }

SQLUnit分组和连接模式选择没有任何关系，连接模式的选择只取决于maxConnectionsSizePerQuery和SQLUnit数量的大小关系，maxConnectionsSizePerQuery代表了一个数据源一次查询允许的最大连接数。

  • 当maxConnectionsSizePerQuery<sqlunit数量时，意味着无法做到每个sqlunit独享一个连接，需要直接查询出结果集至内存中；< li="">

  • 当maxConnectionsSizePerQuery>=SQLUnit数量时，意味着可以支持每个SQLUnit独享一个连接，可以通过ResultSet游标下移的方式查询结果集。

不过maxConnectionsSizePerQuery默认值为1，所以当一条SQL需要路由至多张表时（即有多个SQLUnit）会采用连接限制，当路由至单表时是内存限制模式。

图片

为了避免产生数据库连接死锁问题，在内存限制模式时，Sharding-JDBC通过锁住数据源对象一次性创建出本条SQL需要的所有数据库连接。连接限制模式下，各连接一次性查出各自的结果，不会出现多连接相互等待的情况，因此不会发生死锁，而内存限制模式通过游标读取结果集，需要多条连接去查询不同的表做合并，如果不一次性拿到所有需要的连接，则可能存在连接相互等待的情况造成死锁。可以参照官方文档中执行引擎相关例子。

  • 不同连接模式创建连接

  1. private List<Connection> createConnections(final String dataSourceName, final ConnectionMode connectionMode, final DataSource dataSource, final int connectionSize) throws SQLException {
  2.     if (1 == connectionSize) {
  3.         Connection connection = createConnection(dataSourceName, dataSource);
  4.         replayMethodsInvocation(connection);
  5.         return Collections.singletonList(connection);
  6.     }
  7.     if (ConnectionMode.CONNECTION_STRICTLY == connectionMode) {
  8.         return createConnections(dataSourceName, dataSource, connectionSize);
  9.     }
  10.     //内存限制模式加锁 一次性获取所有的连接
  11.     synchronized (dataSource) {
  12.         return createConnections(dataSourceName, dataSource, connectionSize);
  13.     }
  14. }

此外，结果集的内存合并和流式合并只在调用JDBC的executeQuery的情况下生效，如果使用execute方式进行查询，都是统一使用流式方式的查询。

  • 查询结果归并对比

  1. org.apache.shardingsphere.shardingjdbc.executor.PreparedStatementExecutor#executeQuery#101 
  2.   org.apache.shardingsphere.shardingjdbc.executor.PreparedStatementExecutor#getQueryResult
  3.     private QueryResult getQueryResult(final Statement statement, final ConnectionMode connectionMode) throws SQLException {
  4.         PreparedStatement preparedStatement = (PreparedStatement) statement;
  5.         ResultSet resultSet = preparedStatement.executeQuery();
  6.         getResultSets().add(resultSet);
  7.         //executeQuery 中根据连接模式选择流式/内存
  8.         return ConnectionMode.MEMORY_STRICTLY == connectionMode ? new StreamQueryResult(resultSet) : new MemoryQueryResult(resultSet);
  9.     }
  10.  
  11. //execute 单独调用getResultSet中只会使用流式合并
  12. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#getResultSet#158
  13.   org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#getQueryResults 
  14.     private List<QueryResult> getQueryResults(final List<ResultSet> resultSets) throws SQLException {
  15.         List<QueryResult> result = new ArrayList<>(resultSets.size());
  16.         for (ResultSet each : resultSets) {
  17.             if (null != each) {
  18.                 result.add(new StreamQueryResult(each));
  19.             }
  20.         }
  21.         return result;
  22.     }

多条连接的执行方式分为串行和并行，在本地事务和XA事务中是串行的方式，其余情况是并行，具体的执行逻辑这里就不再展开了。

  • isHoldTransaction

  1. public boolean isHoldTransaction() {
  2.         return (TransactionType.LOCAL == transactionType && !getAutoCommit()) || (TransactionType.XA == transactionType && isInShardingTransaction());
  3.     }

3.5.2 引擎总结

执行引擎通过maxConnectionsSizePerQuery和同数据源的SQLUnit的数量大小确定连接模式，maxConnectionsSizePerQuery=SQLUnit数量使用内存限制模式，当使用内存限制模式时会通过对数据源对象加锁来保证一次性获取本条SQL需要的连接而避免死锁。在使用executeQuery查询时，处理结果集时会根据连接模式选择流式或者内存合并，但使用execute方法查询，处理结果集只会使用流式合并。

3.6 归并引擎

3.6.1 引擎解析

查询出的结果集需要经过归并引擎归并后才是最终的结果，归并的核心入口在MergeEntry的process方法中，优先处理分片场景的合并，再进行脱敏，只有读写分离的情况下则直接返回TransparentMergedResult，TransparentMergedResult实际上没做合并的额外处理，其内部实现都是完全调用queryResult的实现。

图片

  • 归并逻辑入口

  1. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#mergeQuery#190
  2.  org.apache.shardingsphere.underlying.pluggble.merge.MergeEngine#merge#61
  3.     org.apache.shardingsphere.underlying.merge.MergeEntry#process
  4.     public MergedResult process(final List<QueryResult> queryResults, final SQLStatementContext sqlStatementContext) throws SQLException {
  5.         //分片合并
  6.         Optional<MergedResult> mergedResult = merge(queryResults, sqlStatementContext);
  7.         //脱敏处理
  8.         Optional<MergedResult> result = mergedResult.isPresent() ? Optional.of(decorate(mergedResult.get(), sqlStatementContext)) : decorate(queryResults.get(0), sqlStatementContext);
  9.         //只有读写分离的情况下，orElseGet会不存在，TransparentMergedResult
  10.         return result.orElseGet(() -> new TransparentMergedResult(queryResults.get(0)));
  11.     }

  • TransparentMergedResult

  1. @RequiredArgsConstructor
  2. public final class TransparentMergedResult implements MergedResult {
  3.      
  4.     private final QueryResult queryResult;
  5.      
  6.     @Override
  7.     public boolean next() throws SQLException {
  8.         return queryResult.next();
  9.     }
  10.      
  11.     @Override
  12.     public Object getValue(final int columnIndex, final Class<?> type) throws SQLException {
  13.         return queryResult.getValue(columnIndex, type);
  14.     }
  15.      
  16.     @Override
  17.     public Object getCalendarValue(final int columnIndex, final Class<?> type, final Calendar calendar) throws SQLException {
  18.         return queryResult.getCalendarValue(columnIndex, type, calendar);
  19.     }
  20.      
  21.     @Override
  22.     public InputStream getInputStream(final int columnIndex, final String type) throws SQLException {
  23.         return queryResult.getInputStream(columnIndex, type);
  24.     }
  25.      
  26.     @Override
  27.     public boolean wasNull() throws SQLException {
  28.         return queryResult.wasNull();
  29.     }
  30. }

我们只看分片相关的操作，ResultMergerEngine只有一个实现类ShardingResultMergerEngine，所以只有存在分片情况的时候，上文的第一个merge才会有结果。根据SQL类型的不同选择ResultMerger实现，查询类的合并是最常用也是最复杂的合并。

  • MergeEntry.merge

  1. org.apache.shardingsphere.underlying.merge.MergeEntry#merge
  2.     private Optional<MergedResult> merge(final List<QueryResult> queryResults, final SQLStatementContext sqlStatementContext) throws SQLException {
  3.         for (Entry<BaseRule, ResultProcessEngine> entry : engines.entrySet()) {
  4.             if (entry.getValue() instanceof ResultMergerEngine) {
  5.                 //选择不同类型的 resultMerger
  6.                 ResultMerger resultMerger = ((ResultMergerEngine) entry.getValue()).newInstance(databaseType, entry.getKey(), properties, sqlStatementContext);
  7.                 //归并
  8.                 return Optional.of(resultMerger.merge(queryResults, sqlStatementContext, schemaMetaData));
  9.             }
  10.         }
  11.         return Optional.empty();
  12.     }
  13.  
  14. org.apache.shardingsphere.sharding.merge.ShardingResultMergerEngine#newInstance
  15.     public ResultMerger newInstance(final DatabaseType databaseType, final ShardingRule shardingRule, final ConfigurationProperties properties, final SQLStatementContext sqlStatementContext) {
  16.         if (sqlStatementContext instanceof SelectStatementContext) {
  17.             return new ShardingDQLResultMerger(databaseType);
  18.         }
  19.         if (sqlStatementContext.getSqlStatement() instanceof DALStatement) {
  20.             return new ShardingDALResultMerger(shardingRule);
  21.         }
  22.         return new TransparentResultMerger();
  23.     }

ShardingDQLResultMerger的merge方法就是根据SQL解析结果中包含的token选择合适的归并方式（分组聚合、排序、遍历），归并后的mergedResult统一经过decorate方法进行判断是否需要分页归并，整体处理流程图可以概括如下。

  • 归并方式选择

  1. org.apache.shardingsphere.sharding.merge.dql.ShardingDQLResultMerger#merge
  2.     public MergedResult merge(final List<QueryResult> queryResults, final SQLStatementContext sqlStatementContext, final SchemaMetaData schemaMetaData) throws SQLException {
  3.         if (1 == queryResults.size()) {
  4.             return new IteratorStreamMergedResult(queryResults);
  5.         }
  6.         Map<String, Integer> columnLabelIndexMap = getColumnLabelIndexMap(queryResults.get(0));
  7.         SelectStatementContext selectStatementContext = (SelectStatementContext) sqlStatementContext;
  8.         selectStatementContext.setIndexes(columnLabelIndexMap);
  9.         //分组聚合，排序，遍历
  10.         MergedResult mergedResult = build(queryResults, selectStatementContext, columnLabelIndexMap, schemaMetaData);
  11.         //分页归并
  12.         return decorate(queryResults, selectStatementContext, mergedResult);
  13.     }
  14.  
  15. org.apache.shardingsphere.sharding.merge.dql.ShardingDQLResultMerger#build
  16.     private MergedResult build(final List<QueryResult> queryResults, final SelectStatementContext selectStatementContext,
  17.                                final Map<String, Integer> columnLabelIndexMap, final SchemaMetaData schemaMetaData) throws SQLException {
  18.         if (isNeedProcessGroupBy(selectStatementContext)) {
  19.             //分组聚合归并
  20.             return getGroupByMergedResult(queryResults, selectStatementContext, columnLabelIndexMap, schemaMetaData);
  21.         }
  22.         if (isNeedProcessDistinctRow(selectStatementContext)) {
  23.             setGroupByForDistinctRow(selectStatementContext);
  24.             //分组聚合归并
  25.             return getGroupByMergedResult(queryResults, selectStatementContext, columnLabelIndexMap, schemaMetaData);
  26.         }
  27.         if (isNeedProcessOrderBy(selectStatementContext)) {
  28.             //排序归并
  29.             return new OrderByStreamMergedResult(queryResults, selectStatementContext, schemaMetaData);
  30.         }
  31.         //遍历归并
  32.         return new IteratorStreamMergedResult(queryResults);
  33.     }
  34.  
  35. org.apache.shardingsphere.sharding.merge.dql.ShardingDQLResultMerger#decorate
  36.     private MergedResult decorate(final List<QueryResult> queryResults, final SelectStatementContext selectStatementContext, final MergedResult mergedResult) throws SQLException {
  37.         PaginationContext paginationContext = selectStatementContext.getPaginationContext();
  38.         if (!paginationContext.isHasPagination() || 1 == queryResults.size()) {
  39.             return mergedResult;
  40.         }
  41.         String trunkDatabaseName = DatabaseTypes.getTrunkDatabaseType(databaseType.getName()).getName();
  42.         //根据数据库类型分页归并
  43.         if ("MySQL".equals(trunkDatabaseName) || "PostgreSQL".equals(trunkDatabaseName)) {
  44.             return new LimitDecoratorMergedResult(mergedResult, paginationContext);
  45.         }
  46.         if ("Oracle".equals(trunkDatabaseName)) {
  47.             return new RowNumberDecoratorMergedResult(mergedResult, paginationContext);
  48.         }
  49.         if ("SQLServer".equals(trunkDatabaseName)) {
  50.             return new TopAndRowNumberDecoratorMergedResult(mergedResult, paginationContext);
  51.         }
  52.         return mergedResult;
  53.     }

每种归并方式的作用在官方文档有比较详细的案例，这里就不再重复介绍了。

3.6.2 引擎总结

归并引擎是Sharding-JDBC执行SQL的最后一步，其作用是将多个数节点的结果集组合为一个正确的结果集返回，查询类的归并有分组归并、聚合归并、排序归并、遍历归并、分页归并五种，这五种归并方式并不是互斥的，而是相互组合的。

四、定制开发

在使用Sharding-JDBC过程中，我们发现了一些问题可以改进，比如存量系统数据量到达一定规模而需要分库分表引入Sharding-JDBC时，就会存在两大问题。

一个是存量数据的迁移，这个问题我们可以通过分片算法兼容，前文已经提过分片键的值是不允许更改的，而且SQL如果不包含分片键，如果这个分片键对应的值是递增的（如id，时间等），我们可以设置一个阈值，在分片算法的doSharding中判断分片值与阈值的大小决定将数据路由至旧表或新表，避免数据迁移的麻烦。如果是根据用户id取模分表，而新增的数据无法只通过用户id判断，这时可以考虑采用复合分片算法，将用户id与订单id或者时间等递增的字段同时设置为分片键，根据订单id或时间判断是否是新数据，再根据用户id取模得到路由结果即可。

另一个是Sharding-JDBC语法限制会使得存量SQL面对巨大的改造压力，而实际上业务更关心的是需要分片的表，非分片的表不应该发生改动和影响。实际上，非分片表理论上无需通过解析、路由、重写、合并，为此我们在源码层面对这段逻辑进行了优化，支持跳过部分解析，完全跳过分片路由、重写和合并，尽可能减少Sharding-JDBC对非分片表的语法限制，来减少业务系统的改造压力与风险。

图片

4.1 跳过Sharding语法限制

Sharding-JDBC执行解析路由重写的逻辑都是在BasePrepareEngine中，最终构造ExecutionContext交由执行引擎执行，ExecutionContext中包含sqlStatementContext和executionUnits，非分片表不涉及路由改写，所以其ExecutionUnit我们非常容易手动构造，而查看SQLStatementContext的使用情况，我们发现SQLStatementContext只会影响结果集的合并而不会影响实际的执行，而不分片表也无需进行结果集的合并，整体实现思路如图。

图片

  • ExecutionContext相关对象

  1. public class ExecutionContext {
  2.  
  3.     private final SQLStatementContext sqlStatementContext;
  4.  
  5.     private final Collection<ExecutionUnit> executionUnits = new LinkedHashSet<>();
  6. }
  7.  
  8. public final class ExecutionUnit {
  9.      
  10.     private final String dataSourceName;
  11.      
  12.     private final SQLUnit sqlUnit;
  13. }
  14.  
  15. public final class SQLUnit {
  16.  
  17.     private String sql;
  18.  
  19.     private final List<Object> parameters;
  20.  
  21. }

（1）校验SQL中是否包含分片表：我们是通过正则将SQL中的各个单词分隔成Set，然后再遍历BaseRule判断是否存在分片表。大家可能会奇怪明明解析引擎可以帮我们解析出SQL中的表名，为什么还要自己来解析。因为我们测试的过程中发现，存量业务上的SQL很多在解析阶段就会报错，只能提前判断，当然这种判断方式并不严谨，比如 SELECT order_id FROM t_order_record WHERE order_id=1 AND remarks=' t_order xxx';，配置的分片表t_order时就会存在误判，但这种场景在我们的业务中没有，所以暂时并没有处理。由于这个信息需要在多个对象方法中使用，为了避免修改大量的对象变量和方法入参，而又能方便的透传这个信息，判断的结果我们选择放在ThreadLocal里。

  • RuleContextManager

  1. public final class RuleContextManager {
  2.  
  3.     private static final ThreadLocal<RuleContextManager> SKIP_CONTEXT_HOLDER = ThreadLocal.withInitial(RuleContextManager::new);
  4.  
  5.     /**
  6.      * 是否跳过sharding
  7.      */
  8.     private boolean skipSharding;
  9.  
  10.     /**
  11.      * 是否路由至主库
  12.      */
  13.     private boolean masterRoute;
  14.  
  15.     public static boolean isSkipSharding() {
  16.         return SKIP_CONTEXT_HOLDER.get().skipSharding;
  17.     }
  18.  
  19.     public static void setSkipSharding(boolean skipSharding) {
  20.         SKIP_CONTEXT_HOLDER.get().skipSharding = skipSharding;
  21.     }
  22.  
  23.     public static boolean isMasterRoute() {
  24.  
  25.         return SKIP_CONTEXT_HOLDER.get().masterRoute;
  26.     }
  27.  
  28.     public static void setMasterRoute(boolean masterRoute) {
  29.         SKIP_CONTEXT_HOLDER.get().masterRoute = masterRoute;
  30.     }
  31.  
  32.     public static void clear(){
  33.         SKIP_CONTEXT_HOLDER.remove();
  34.     }
  35.  
  36. }

  • 判断SQL是否包含分片表

  1. org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine#buildSkipContext
  2. // 判断是否可以跳过sharding,构造RuleContextManager的值
  3. private void buildSkipContext(final String sql){
  4.     Set<String> sqlTokenSet = new HashSet<>(Arrays.asList(sql.split("[\\s]")));
  5.         if (CollectionUtils.isNotEmpty(rules)) {
  6.             for (BaseRule baseRule : rules) {
  7.                 //定制方法，ShardingRule实现，判断sqlTokenSet是否包含逻辑表即可
  8.                 if(baseRule.hasContainShardingTable(sqlTokenSet)){
  9.                     RuleContextManager.setSkipSharding(false);
  10.                     break;
  11.                 }else {
  12.                     RuleContextManager.setSkipSharding(true);
  13.                 }
  14.             }
  15.         }
  16. }
  17.  
  18. org.apache.shardingsphere.core.rule.ShardingRule#hasContainShardingTable
  19. public Boolean hasContainShardingTable(Set<String> sqlTokenSet) {
  20.       //logicTableNameList通过遍历TableRule可以得到
  21.        for (String logicTable : logicTableNameList) {
  22.             if (sqlTokenSet.contains(logicTable)) {
  23.                 return true;
  24.             }
  25.         }
  26.         return false;
  27.     }

（2）跳过解析路由：通过RuleContextManager中的skipSharding判断是否需要跳过Sharding解析路由，但为了兼容读写分离的场景，我们还需要知道这条SQL应该走主库还是从库，走主库的场景在后面强制路由主库部分有说明，SQL走主库实际上只有两种情况，一种是非SELECT语句，另一种就是SELECT语句带锁，如SELECT...FOR UPDATE，因此整体实现的步骤如下：

  • 如果标记了跳过Sharding且不为select语句，直接返回SkipShardingStatement，单独构造一个SkipShardingStatement的目的是为了能利用解析引擎中的缓存，缓存中不能放入null值。

  • 如果是select语句需要继续解析，判断是否有锁后直接返回，避免后续解析造成语法不兼容，这里也曾尝试用反射获取lockClause来判断是否包含锁，但最终没有成功。

  • ShardingRouteDecorator根据RuleContextManager.isSkipSharding判断是否跳过路由。

  • 跳过解析路由

  1. public class SkipShardingStatement implements SQLStatement{
  2.     @Override
  3.     public int getParameterCount() {
  4.         return 0;
  5.     }
  6. }
  7.  
  8. org.apache.shardingsphere.sql.parser.SQLParserEngine#parse0
  9.     private SQLStatement parse0(final String sql, final boolean useCache) {
  10.         if (useCache) {
  11.             Optional<SQLStatement> cachedSQLStatement = cache.getSQLStatement(sql);
  12.             if (cachedSQLStatement.isPresent()) {
  13.                 return cachedSQLStatement.get();
  14.             }
  15.         }
  16.         ParseTree parseTree = new SQLParserExecutor(databaseTypeName, sql).execute().getRootNode();
  17.         /**
  18.          * 跳过sharding 需要判断是否需要路由至主库 如果不是select语句直接跳过
  19.          * 是select语句则需要通过继续解析判断是否有锁
  20.          */
  21.         SQLStatement result ;
  22.         if(RuleContextManager.isSkipSharding()&&!VisitorRule.SELECT.equals(VisitorRule.valueOf(parseTree.getClass()))){
  23.             RuleContextManager.setMasterRoute(true);
  24.             result = new SkipShardingStatement();
  25.         }else {
  26.             result = (SQLStatement) ParseTreeVisitorFactory.newInstance(databaseTypeName, VisitorRule.valueOf(parseTree.getClass())).visit(parseTree);
  27.         }
  28.         if (useCache) {
  29.             cache.put(sql, result);
  30.         }
  31.         return result;
  32.     }
  33.  
  34. org.apache.shardingsphere.sql.parser.mysql.visitor.impl.MySQLDMLVisitor#visitSelectClause
  35.     public ASTNode visitSelectClause(final SelectClauseContext ctx) {
  36.         SelectStatement result = new SelectStatement();
  37.         // 跳过sharding 只需要判断是否有锁来决定是否路由至主库即可
  38.         if(RuleContextManager.isSkipSharding()){
  39.             if (null != ctx.lockClause()) {
  40.                 result.setLock((LockSegment) visit(ctx.lockClause()));
  41.                 RuleContextManager.setMasterRoute(true);
  42.             }
  43.             return result;
  44.         }
  45.         //...后续解析
  46.     }
  47.  
  48. org.apache.shardingsphere.underlying.route.DataNodeRouter#createRouteContext
  49.     private RouteContext createRouteContext(final String sql, final List<Object> parameters, final boolean useCache) {
  50.         SQLStatement sqlStatement = parserEngine.parse(sql, useCache);
  51.         //如果需要跳过sharding 不进行后续的解析直接返回
  52.         if (RuleContextManager.isSkipSharding()) {
  53.             return new RouteContext(sqlStatement, parameters, new RouteResult());
  54.         }
  55.         //...解析
  56.     }
  57.  
  58. org.apache.shardingsphere.sharding.route.engine.ShardingRouteDecorator#decorate
  59.     public RouteContext decorate(final RouteContext routeContext, final ShardingSphereMetaData metaData, final ShardingRule shardingRule, final ConfigurationProperties properties) {
  60.         // 跳过sharding路由
  61.         if(RuleContextManager.isSkipSharding()){
  62.             return routeContext;
  63.         }
  64.         //...路由
  65.     ｝

（3）手动构造ExecutionUnit：ExecutionUnit中我们需要确定的内容就是datasourceName，这里我们认为跳过Sharding的SQL最终执行的库一定只有一个。如果只是跳过Sharding的情况，直接从元数据中获取数据源名称即可，如果存在读写分离的情况，主从路由的结果也一定是唯一的。创建完ExecutionUnit直接放入ExecutionContext返回即可，从而跳过后续的改写逻辑。

  • 手动构造ExecutionUnit

  1. public ExecutionContext prepare(final String sql, final List<Object> parameters) {
  2.     List<Object> clonedParameters = cloneParameters(parameters);
  3.     // 判断是否可以跳过sharding,构造RuleContextManager的值
  4.     buildSkipContext(sql);  
  5.     RouteContext routeContext = executeRoute(sql, clonedParameters);
  6.     ExecutionContext result = new ExecutionContext(routeContext.getSqlStatementContext());
  7.     // 跳过sharding的sql最后的路由结果一定只有一个库
  8.     if(RuleContextManager.isSkipSharding()){
  9.         log.debug("可以跳过sharding的场景 {}", sql);
  10.         if(!Objects.isNull(routeContext.getRouteResult())){
  11.             Collection<String> allInstanceDataSourceNames = this.metaData.getDataSources().getAllInstanceDataSourceNames();
  12.             int routeUnitsSize = routeContext.getRouteResult().getRouteUnits().size();
  13.             /*
  14.              * 1. 没有读写分离的情况下  跳过sharding路由会导致routeUnitsSize为0 此时需要判断数据源数量是否为1
  15.              * 2. 读写分离情况下 只会路由至具体的主库或从库 routeUnitsSize数量应该为1
  16.              */
  17.             if(!(routeUnitsSize == 0 && allInstanceDataSourceNames.size()==1)|| routeUnitsSize>1){
  18.                 throw new ShardingSphereException("可以跳过sharding,但是路由结果不唯一,SQL= %s ,routeUnits= %s ",sql, routeContext.getRouteResult().getRouteUnits());
  19.             }
  20.             Collection<String> actualDataSourceNames = routeContext.getRouteResult().getActualDataSourceNames();
  21.             // 手动创建执行单元
  22.             String datasourceName = CollectionUtils.isEmpty(actualDataSourceNames)? allInstanceDataSourceNames.iterator().next():actualDataSourceNames.iterator().next();
  23.             ExecutionUnit executionUnit = new ExecutionUnit(datasourceName, new SQLUnit(sql, clonedParameters));
  24.             result.getExecutionUnits().add(executionUnit);
  25.             //标记该结果需要跳过
  26.             result.setSkipShardingScenarioFlag(true);
  27.         }
  28.     }else {
  29.         result.getExecutionUnits().addAll(executeRewrite(sql, clonedParameters, routeContext));
  30.     }
  31.     if (properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SHOW)) {
  32.         SQLLogger.logSQL(sql, properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SIMPLE), result.getSqlStatementContext(), result.getExecutionUnits());
  33.     }
  34.     return result;
  35. }

（4）跳过合并：跳过查询结果的合并和影响行数计算的合并，注意ShardingPreparedStatement和ShardingStatement都需要跳过

  • 跳过合并

  1. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#executeQuery
  2.     public ResultSet executeQuery() throws SQLException {
  3.         ResultSet result;
  4.         try {
  5.             clearPrevious();
  6.             prepare();
  7.             initPreparedStatementExecutor();
  8.             List<QueryResult> queryResults = preparedStatementExecutor.executeQuery();
  9.             List<ResultSet> resultSets = preparedStatementExecutor.getResultSets();
  10.         // 定制开发，不分片跳过合并
  11.             if(executionContext.isSkipShardingScenarioFlag()){
  12.                 return CollectionUtils.isNotEmpty(resultSets) ? resultSets.get(0) : null;
  13.             }
  14.             MergedResult mergedResult = mergeQuery(queryResults);
  15.             result = new ShardingResultSet(resultSets, mergedResult, this, executionContext);
  16.         } finally {
  17.             clearBatch();
  18.         }
  19.         currentResultSet = result;
  20.         return result;
  21.     }
  22. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#getResultSet
  23.     public ResultSet getResultSet() throws SQLException {
  24.         if (null != currentResultSet) {
  25.             return currentResultSet;
  26.         }
  27.         List<ResultSet> resultSets = getResultSets();
  28.         // 定制开发，不分片跳过合并
  29.         if(executionContext.isSkipShardingScenarioFlag()){
  30.             return CollectionUtils.isNotEmpty(resultSets) ? resultSets.get(0) : null;
  31.         }
  32.  
  33.         if (executionContext.getSqlStatementContext() instanceof SelectStatementContext || executionContext.getSqlStatementContext().getSqlStatement() instanceof DALStatement) {
  34.             MergedResult mergedResult = mergeQuery(getQueryResults(resultSets));
  35.             currentResultSet = new ShardingResultSet(resultSets, mergedResult, this, executionContext);
  36.         }
  37.         return currentResultSet;
  38.     }
  39. org.apache.shardingsphere.shardingjdbc.jdbc.core.statement.ShardingPreparedStatement#isAccumulate
  40.     public boolean isAccumulate() {
  41.         //定制开发，不分片跳过计算
  42.         if(executionContext.isSkipShardingScenarioFlag()){
  43.             return false;
  44.         }
  45.         return !connection.getRuntimeContext().getRule().isAllBroadcastTables(executionContext.getSqlStatementContext().getTablesContext().getTableNames());
  46.     }

（5）清空RuleContextManager：查看一下Sharding-JDBC其他ThreadLocal的清空位置，对应的清空RuleContextManager就好。

  • 清空ThreadLocal

  1. org.apache.shardingsphere.shardingjdbc.jdbc.adapter.AbstractConnectionAdapter#close
  2. public final void close() throws SQLException {
  3.         closed = true;
  4.         MasterVisitedManager.clear();
  5.         TransactionTypeHolder.clear();
  6.         RuleContextManager.clear();
  7.         int connectionSize = cachedConnections.size();
  8.         try {
  9.             forceExecuteTemplateForClose.execute(cachedConnections.entries(), cachedConnections -> cachedConnections.getValue().close());
  10.         } finally {
  11.             cachedConnections.clear();
  12.             rootInvokeHook.finish(connectionSize);
  13.         }
  14.     }

举个例子，比如Sharding-JDBC本身是不支持INSERT INTO tbl_name (col1, col2, …) SELECT col1, col2, … FROM tbl_name WHERE col3 = ?    这种语法的，会报空指针异常。

图片

经过我们上述改造验证后，非分片表是可以跳过语法限制执行如下的SQL的。

图片

通过该功能的实现，业务可以更关注与分片表的SQL改造，而无需担心引入Sharding-JDBC造成所有SQL的验证改造，大幅减少改造成本和风险。

4.2 强制路由主库

Sharding-JDBC可以通过配置主从库数据源方便的实现读写分离的功能，但使用读写分离就必须面对主从延迟和从库失联的痛点，针对这一问题，我们实现了强制路由主库的动态配置，当主从延迟过大或从库失联时，通过修改配置来实现SQL语句强制走主库的不停机路由切换。

后面会说明了配置的动态生效的实现方式，这里只说明强制路由主库的实现，我们直接使用前文的RuleContextManager即可，在主从路由引擎里判断下是否开启了强制主库路由。

  • MasterSlaveRouteDecorator.decorate改造

  1. org.apache.shardingsphere.masterslave.route.engine.MasterSlaveRouteDecorator#decorate
  2.     public RouteContext decorate(final RouteContext routeContext, final ShardingSphereMetaData metaData, final MasterSlaveRule masterSlaveRule, final ConfigurationProperties properties) {
  3.         /**
  4.          * 如果配置了强制主库 MasterVisitedManager设置为true
  5.          * 后续isMasterRoute中会保证路由至主库
  6.          */
  7.         if(properties.<Boolean>getValue(ConfigurationPropertyKey.MASTER_ROUTE_ONLY)){
  8.             MasterVisitedManager.setMasterVisited();
  9.         }
  10.         //...路由逻辑
  11.         return routeContext;
  12.     }

为了兼容之前跳过Sharding的功能，我们需要同步修改下isMasterRoute方法，如果是跳过了Sharding路由需要通过RuleContextManager来判断是否走主库。

  • isMasterRoute改造

  1. org.apache.shardingsphere.masterslave.route.engine.impl.MasterSlaveDataSourceRouter#isMasterRoute
  2.     private boolean isMasterRoute(final SQLStatement sqlStatement) {
  3.         if(sqlStatement instanceof SkipShardingStatement){
  4.             // 优先以MasterVisitedManager中的值为准
  5.             return MasterVisitedManager.isMasterVisited()|| RuleContextManager.isMasterRoute();
  6.         }
  7.         return containsLockSegment(sqlStatement) || !(sqlStatement instanceof SelectStatement) || MasterVisitedManager.isMasterVisited() || HintManager.isMasterRouteOnly();
  8.     }

当然，更理想的状况是通过监控主从同步延迟和数据库拨测，当超过阈值时或从库失联时直接自动修改配置中心的库，实现自动切换主库，减少业务故障时间和运维压力。

4.3 配置动态生效

Sharding-JDBC中的ConfigurationPropertyKey中提供了许多配置属性，而Sharding-JDBCB并没有为这些配置提供在线修改的方法，而在实际的应用场景中，像SQL_SHOW这样控制SQL打印的开关配置，我们更希望能够在线修改配置值来控制SQL日志的打印，而不是修改完配置再重启服务。

以SQL打印为例，BasePrepareEngine中存在ConfigurationProperties对象，通过调用getValue方法来获取SQL_SHOW的值。

  • SQL 打印

  1. org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine#prepare
  2.     /**
  3.      * Prepare to execute.
  4.      *
  5.      * @param sql SQL
  6.      * @param parameters SQL parameters
  7.      * @return execution context
  8.      */
  9.     public ExecutionContext prepare(final String sql, final List<Object> parameters) {
  10.         List<Object> clonedParameters = cloneParameters(parameters);
  11.         RouteContext routeContext = executeRoute(sql, clonedParameters);
  12.         ExecutionContext result = new ExecutionContext(routeContext.getSqlStatementContext());
  13.         result.getExecutionUnits().addAll(executeRewrite(sql, clonedParameters, routeContext));
  14.         //sql打印
  15.         if (properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SHOW)) {
  16.             SQLLogger.logSQL(sql, properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SIMPLE), result.getSqlStatementContext(), result.getExecutionUnits());
  17.         }
  18.         return result;
  19.     }

ConfigurationProperties继承了抽象类TypedProperties，其getValue方法就是根据key获取对应的配置值，因此我们直接在TypedProperties中实现刷新缓存中的配置值的方法。

  • TypedProperties刷新配置

  1. public abstract class TypedProperties<E extends Enum & TypedPropertyKey> {
  2.      
  3.     private static final String LINE_SEPARATOR = System.getProperty("line.separator");
  4.      
  5.     @Getter
  6.     private final Properties props;
  7.      
  8.     private final Map<E, TypedPropertyValue> cache;
  9.      
  10.     public TypedProperties(final Class<E> keyClass, final Properties props) {
  11.         this.props = props;
  12.         cache = preload(keyClass);
  13.     }
  14.      
  15.     private Map<E, TypedPropertyValue> preload(final Class<E> keyClass) {
  16.         E[] enumConstants = keyClass.getEnumConstants();
  17.         Map<E, TypedPropertyValue> result = new HashMap<>(enumConstants.length, 1);
  18.         Collection<String> errorMessages = new LinkedList<>();
  19.         for (E each : enumConstants) {
  20.             TypedPropertyValue value = null;
  21.             try {
  22.                 value = new TypedPropertyValue(each, props.getOrDefault(each.getKey(), each.getDefaultValue()).toString());
  23.             } catch (final TypedPropertyValueException ex) {
  24.                 errorMessages.add(ex.getMessage());
  25.             }
  26.             result.put(each, value);
  27.         }
  28.         if (!errorMessages.isEmpty()) {
  29.             throw new ShardingSphereConfigurationException(Joiner.on(LINE_SEPARATOR).join(errorMessages));
  30.         }
  31.         return result;
  32.     }
  33.      
  34.     /**
  35.      * Get property value.
  36.      *
  37.      * @param key property key
  38.      * @param <T> class type of return value
  39.      * @return property value
  40.      */
  41.     @SuppressWarnings("unchecked")
  42.     public <T> T getValue(final E key) {
  43.         return (T) cache.get(key).getValue();
  44.     }
  45.  
  46.     /**
  47.      * vivo定制改造方法 refresh property value.
  48.      * @param key property key
  49.      * @param value property value
  50.      * @return 更新配置是否成功
  51.      */
  52.     public boolean refreshValue(String key, String value){
  53.         //获取配置类支持的配置项
  54.         E[] enumConstants = targetKeyClass.getEnumConstants();
  55.         for (E each : enumConstants) {
  56.             //遍历新的值
  57.             if(each.getKey().equals(key)){
  58.                 try {
  59.                     //空白value认为无效,取默认值
  60.                     if(!StringUtils.isBlank(value)){
  61.                         value = each.getDefaultValue();
  62.                     }
  63.                     //构造新属性
  64.                     TypedPropertyValue typedPropertyValue = new TypedPropertyValue(each, value);
  65.                     //替换缓存
  66.                     cache.put(each, typedPropertyValue);
  67.                     //原始属性也替换下,有可能会通过RuntimeContext直接获取Properties
  68.                     props.put(key,value);
  69.                     return true;
  70.                 } catch (final TypedPropertyValueException ex) {
  71.                     log.error("refreshValue error. key={} , value={}", key, value, ex);
  72.                 }
  73.             }
  74.         }
  75.         return false;
  76.     }
  77. }

实现了刷新方法后，我们还需要将该方法一步步暴露至一个外部可以调用的类中，以便在服务监听配置的方法中，能够调用这个刷新方法。ConfigurationProperties直接在BasePrepareEngine的构造函数中传入，我们通过构造函数逐步反推最外层的这一对象调用来源，最终可以定位到在AbstractDataSourceAdapter中的getRuntimeContext()方法中可以获取到这个配置，而这个就是Sharding-JDBC实现的JDBC中Datasource接口的抽象类，我们直接在这个类中调用刚刚实现的refreshValue方法，剩下的就是监听配置，通过自己实现的AbstractDataSourceAdapter来调用这个方法就好了。

图片

通过这一功能，我们可以方便的控制一些开关属性的在线修改，如SQL打印、强制路由主库等，业务无需重启服务即可做到配置的动态生效。

4.4 批量update语法支持

业务中存在使用foreach标签来批量update的语句，这种SQL在Sharding-JDBC中无法被正确路由，只会路由第一组参数，后面的无法被路由改写，原因是解析引擎无法将语句拆分解析。

  • 批量update样例

  1. <update id="batchUpdate">
  2.         <foreach collection="orderList" item="item">
  3.                update t_order set
  4.                status = 1,
  5.                updated_by = #{item.updatedBy}
  6.                WHERE created_by = #{item.createdBy};
  7.         </foreach>
  8.     </update>

图片

图片

我们通过将批量update按照;拆分为多个语句，然后分别路由，最后手动汇总路有结果生成执行单元。

为了能正确重写SQL，批量update拆分后的语句需要完全一样，这样就不能使用动态拼接set条件，而是使用ifnull语法或者字段值不发生变化时也将原来的值放入set中，只不过set前后的值保持一致，整体思路与实现如下。

图片

  • prepareBatch实现

  1.  
  2. org.apache.shardingsphere.underlying.pluggble.prepare.BasePrepareEngine#prepareBatch
  3.    private ExecutionContext prepareBatch(List<String> splitSqlList, final List<Object> allParameters) {
  4.        //SQL去重
  5.        List<String> sqlList = splitSqlList.stream().distinct().collect(Collectors.toList());
  6.        if (sqlList.size() > 1) {
  7.            throw new ShardingSphereException("不支持多条SQL,请检查SQL," + sqlList.toString());
  8.        }
  9.        //以第一条SQL为标准
  10.        String sql = sqlList.get(0);
  11.        //所有的执行单元
  12.        Collection<ExecutionUnit> globalExecutionUnitList = new ArrayList<>();
  13.        //初始化最后的执行结果
  14.        ExecutionContext executionContextResult = null;
  15.        //根据所有参数数量和SQL语句数量 计算每组参数的数量
  16.        int eachSqlParameterCount = allParameters.size() / splitSqlList.size();
  17.        //平均分配每条SQL的参数
  18.        List<List<Object>> eachSqlParameterListList = Lists.partition(allParameters, eachSqlParameterCount);
  19.        for (List<Object> eachSqlParameterList : eachSqlParameterListList) {
  20.            //每条SQL参数不同 需要根据参数路由不同的结果  实际的SqlStatementContext 是一致的
  21.            RouteContext routeContext = executeRoute(sql, eachSqlParameterList);
  22.            //由于SQL一样  实际的SqlStatementContext 是一致的 只需初始化一次
  23.            if (executionContextResult == null) {
  24.                executionContextResult = new ExecutionContext(routeContext.getSqlStatementContext());
  25.            }
  26.            globalExecutionUnitList.addAll(executeRewrite(sql, eachSqlParameterList, routeContext));
  27.        }
  28.        //排序打印日志
  29.        executionContextResult.getExtendMap().put(EXECUTION_UNIT_LIST, globalExecutionUnitList.stream().sorted(Comparator.comparing(ExecutionUnit::getDataSourceName)).collect(Collectors.toList()));
  30.        if (properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SHOW)) {
  31.            SQLLogger.logSQL(sql, properties.<Boolean>getValue(ConfigurationPropertyKey.SQL_SIMPLE),
  32.                    executionContextResult.getSqlStatementContext(), (Collection<ExecutionUnit>) executionContextResult.getExtendMap().get(EXECUTION_UNIT_LIST));
  33.        }
  34.        return executionContextResult;
  35.    }

这里我们在ExecutionContext单独构造了一个了ExtendMap来存放ExecutionUnit，原因是ExecutionContext中的executionUnits是HashSet，而判断ExecutionUnit中的SqlUnit只会根据SQL去重，批量update的SQL是一致的，但parameters不同，为了不影响原有的逻辑，单独使用了另外的变量来存放。

  • ExecutionContext改造

  1. @RequiredArgsConstructor
  2. @Getter
  3. public class ExecutionContext {
  4.  
  5.     private final SQLStatementContext sqlStatementContext;
  6.  
  7.     private final Collection<ExecutionUnit> executionUnits = new LinkedHashSet<>();
  8.  
  9.     /**
  10.      * 自定义扩展变量
  11.      */
  12.     private final Map<ExtendEnum,Object> extendMap = new HashMap<>();
  13.  
  14.     /**
  15.      * 定制扩展，是否可以跳过分片逻辑
  16.      */
  17.     @Setter
  18.     private boolean skipShardingScenarioFlag = false;
  19. }
  20.  
  21. @RequiredArgsConstructor
  22. @Getter
  23. @EqualsAndHashCode
  24. @ToString
  25. public final class ExecutionUnit {
  26.      
  27.     private final String dataSourceName;
  28.      
  29.     private final SQLUnit sqlUnit;
  30. }
  31.  
  32. @AllArgsConstructor
  33. @RequiredArgsConstructor
  34. @Getter
  35. @Setter
  36. //根据SQL判断是否相等
  37. @EqualsAndHashCode(of = { "sql" })
  38. @ToString
  39. public final class SQLUnit {
  40.  
  41.     private String sql;
  42.  
  43.     private final List<Object> parameters;
  44.  
  45. }

我们还需要改造下执行方法，在初始化执行器的时候，判断下ExtendMap中存在我们自定义的EXECUTION_UNIT_LIST是否存在，存在则使用生成InputGroup，同一个数据源下的ExecutionUnit会被放入同一个InputGroup中。

  • InputGroup改造

  1. org.apache.shardingsphere.shardingjdbc.executor.PreparedStatementExecutor#init
  2.     public void init(final ExecutionContext executionContext) throws SQLException {
  3.         setSqlStatementContext(executionContext.getSqlStatementContext());
  4.         //兼容批量update 分库分表后同一张表的情况 判断是否存在EXECUTION_UNIT_LIST 存在则使用未去重的List进行后续的操作
  5.         if (MapUtils.isNotEmpty(executionContext.getExtendMap())){
  6.             Collection<ExecutionUnit> executionUnitCollection = (Collection<ExecutionUnit>) executionContext.getExtendMap().get(EXECUTION_UNIT_LIST);
  7.             if(CollectionUtils.isNotEmpty(executionUnitCollection)){
  8.                 getInputGroups().addAll(obtainExecuteGroups(executionUnitCollection));
  9.             }
  10.         }else {
  11.             getInputGroups().addAll(obtainExecuteGroups(executionContext.getExecutionUnits()));
  12.         }
  13.         cacheStatements();
  14.     }

改造完成后，批量update中的每条SQL都可以被正确路由执行。

图片

4.5 ShardingCondition去重

当where语句包括多个or条件时，而or条件不包含分片键时，会造成createShardingConditions方法生成重复的分片条件，导致重复调用doSharding方法。

如SELECT * FROM t_order  WHERE created_by = ? and (   (status = ?) or  (status = ?) or  (status = ?) )这种SQL，存在三个or条件，分片键是created_by ，实际产生的shardingCondition会是三个一样的值，并会调用三次doSharding的方法。虽然实际执行还是只有一次（批量update那里说明过执行单元会去重），但为了减少方法的重复调用，我们还是对这里做了一次去重。

图片

图片

去重的方法也比较简单粗暴，我们对ListRouteValue和RangeRouteValue添加了@EqualsAndHashCode注解，然后在WhereClauseShardingConditionEngine的createShardingConditions方法返回最终结果前加一次去重，从而避免生成重复的shardingCondition造成doSharding方法的重复调用。

  • createShardingConditions去重

  1. org.apache.shardingsphere.sharding.route.engine.condition.engine.WhereClauseShardingConditionEngine#createShardingConditions
  2.     private Collection<ShardingCondition> createShardingConditions(final SQLStatementContext sqlStatementContext, final Collection<AndPredicate> andPredicates, final List<Object> parameters) {
  3.         Collection<ShardingCondition> result = new LinkedList<>();
  4.         for (AndPredicate each : andPredicates) {
  5.             Map<Column, Collection<RouteValue>> routeValueMap = createRouteValueMap(sqlStatementContext, each, parameters);
  6.             if (routeValueMap.isEmpty()) {
  7.                 return Collections.emptyList();
  8.             }
  9.             result.add(createShardingCondition(routeValueMap));
  10.         }
  11.         //去重
  12.         Collection<ShardingCondition> distinctResult = result.stream().distinct().collect(Collectors.toCollection(LinkedList::new));
  13.         return distinctResult;
  14.     }

4.6  全路由校验

分片表的SQL中如果没有携带分片键（或者带上了分片键结果没有被正确解析）将会导致全路由，产生性能问题，而这种SQL并不会报错，这就导致在实际的业务改造中，开发和测试很难保证百分百改造彻底。为此，我们在源码层面对这种情况做了额外的校验，当产生全路由，也就是ShardingConditions为空时，主动抛出异常，从而方便开发和测试能够快速发现全路由SQL。

实现方式也比较简单，校验下ShardingConditions是否为空即可，只不过需要额外兼容下Hint策略ShardingConditions始终为空的特殊情况。

  • 全路由校验

  1. org.apache.shardingsphere.sharding.route.engine.ShardingRouteDecorator#decorate
  2. public RouteContext decorate(final RouteContext routeContext, final ShardingSphereMetaData metaData, final ShardingRule shardingRule, final ConfigurationProperties properties) {
  3.         //省略...
  4.         //获取 ShardingConditions
  5.         ShardingConditions shardingConditions = getShardingConditions(parameters, sqlStatementContext, metaData.getSchema(), shardingRule);
  6.         boolean hintAlgorithm = isHintAlgorithm(sqlStatementContext, shardingRule);
  7.         //判断是否允许全路由
  8.         if (!properties.<Boolean>getValue(ConfigurationPropertyKey.ALLOW_EMPTY_SHARDING_CONDITIONS)) {
  9.             //如果不是Hint算法
  10.             if(!isHintAlgorithm(sqlStatementContext, shardingRule)){
  11.                 /** 如果是DML语句  则可能有两种情况 这两种情况是根据getShardingConditions方法的内部逻辑而来的
  12.                  *  一种是非插入语句  shardingConditions.getConditions()为空即可
  13.                  *  一种是插入语句 插入语句shardingConditions.getConditions()不会为空  但是ShardingCondition的routeValues是空的
  14.                  */
  15.                 if (sqlStatementContext.getSqlStatement() instanceof DMLStatement) {
  16.                     if(shardingConditions.getConditions().isEmpty()) {
  17.                         throw new ShardingSphereException("SQL不包含分库分表键,请检查SQL");
  18.                     }else {
  19.                         if (sqlStatementContext instanceof InsertStatementContext) {
  20.                             List<ShardingCondition> routeValuesNotEmpty = shardingConditions.getConditions().stream().filter(r -> CollectionUtils.isNotEmpty(r.getRouteValues())).collect(Collectors.toList());
  21.                             if(CollectionUtils.isEmpty(routeValuesNotEmpty)){
  22.                                 throw new ShardingSphereException("SQL不包含分库分表键,请检查SQL");
  23.                             }
  24.                         }
  25.                     }
  26.                 }
  27.             }
  28.         }
  29.         boolean needMergeShardingValues = isNeedMergeShardingValues(sqlStatementContext, shardingRule);
  30.         //省略...
  31.         return new RouteContext(sqlStatementContext, parameters, routeResult);
  32.     }
  33.  
  34. private boolean isHintAlgorithm(final SQLStatementContext sqlStatementContext, final ShardingRule shardingRule) {
  35.         // 场景a 全局默认策略是否使用强制路由策略
  36.         if(shardingRule.getDefaultDatabaseShardingStrategy() instanceof HintShardingStrategy
  37.                 || shardingRule.getDefaultTableShardingStrategy() instanceof HintShardingStrategy){
  38.             return true;
  39.         }
  40.         for (String each : sqlStatementContext.getTablesContext().getTableNames()) {
  41.             Optional<TableRule> tableRule = shardingRule.findTableRule(each);
  42.             //场景b 指定表是否使用强制路由策略
  43.             if (tableRule.isPresent() && (shardingRule.getDatabaseShardingStrategy(tableRule.get()) instanceof HintShardingStrategy
  44.                     || shardingRule.getTableShardingStrategy(tableRule.get()) instanceof HintShardingStrategy)) {
  45.                 return true;
  46.             }
  47.         }
  48.         return false;
  49.     }

当然这块功能也可以在完善些，比如对分片路由结果中的数据源数量进行校验，从而避免跨库操作，我们这边没有实现也就不再赘述了。

4.7 组件封装

业务接入Sharding-JDBC的步骤是一样的，都需要通过Java创建数据源和配置对象或者使用SpringBoot进行配置，存在一定的熟悉成本和重复开发的问题，为此我们也对定制开发版本的Sharding-JDBC封装了一个公共组件，从而简化业务配置，减少重复开发，提升业务的开发效率，具体功能可见下。这块没有涉及源码的改造，只是在定制版本上包装的一个公共组件。

  • 提供了默认的数据源与连接池配置

  • 简化分库分表配置，业务配置逻辑表名和后缀，组件拼装行表达式和actual-data-nodes

  • 封装常用的分片算法（时间、业务字段值等），

  • 统一的配置监听与动态修改（SQL打印、强制主从切换等）

开源Sharding-JDBC配置

  1. //数据源名称
  2. spring.shardingsphere.datasource.names=ds0,ds1
  3. //ds0配置
  4. spring.shardingsphere.datasource.ds0.type=org.apache.commons.dbcp.BasicDataSource
  5. spring.shardingsphere.datasource.ds0.driver-class-name=com.mysql.jdbc.Driver
  6. spring.shardingsphere.datasource.ds0.url=jdbc:mysql://localhost:3306/ds0
  7. spring.shardingsphere.datasource.ds0.username=root
  8. spring.shardingsphere.datasource.ds0.password=
  9. //ds1配置
  10. spring.shardingsphere.datasource.ds1.type=org.apache.commons.dbcp.BasicDataSource
  11. spring.shardingsphere.datasource.ds1.driver-class-name=com.mysql.jdbc.Driver
  12. spring.shardingsphere.datasource.ds1.url=jdbc:mysql://localhost:3306/ds1
  13. spring.shardingsphere.datasource.ds1.username=root
  14. spring.shardingsphere.datasource.ds1.password=
  15. //分表规则
  16. spring.shardingsphere.sharding.tables.t_order.actual-data-nodes=ds$->{0..1}.t_order$->{0..1}
  17. spring.shardingsphere.sharding.tables.t_order.table-strategy.inline.sharding-column=order_id
  18. spring.shardingsphere.sharding.tables.t_order.table-strategy.inline.algorithm-expression=t_order$->{order_id % 2}
  19. spring.shardingsphere.sharding.tables.t_order_item.actual-data-nodes=ds$->{0..1}.t_order_item$->{0..1}
  20. spring.shardingsphere.sharding.tables.t_order_item.table-strategy.inline.sharding-column=order_id
  21. spring.shardingsphere.sharding.tables.t_order_item.table-strategy.inline.algorithm-expression=t_order_item$->{order_id % 2}
  22. //默认分库规则
  23. spring.shardingsphere.sharding.default-database-strategy.inline.sharding-column=user_id
  24. spring.shardingsphere.sharding.default-database-strategy.inline.algorithm-expression=ds$->{user_id % 2}

组件简化配置

  1. //数据源名称
  2. vivo.it.sharding.datasource.names = ds0,ds1
  3. //ds0配置
  4. vivo.it.sharding.datasource.ds0.url = jdbc:mysql://localhost:3306/ds1
  5. vivo.it.sharding.datasource.ds0.username = root
  6. vivo.it.sharding.datasource.ds0.password =
  7. //ds1配置
  8. vivo.it.sharding.datasource.ds1.url = jdbc:mysql://localhost:3306/ds1
  9. vivo.it.sharding.datasource.ds1.username = root
  10. vivo.it.sharding.datasource.ds1.password =
  11. //分表规则
  12. vivo.it.sharding.table.rule.config = [{"logicTable":"t_order,t_order_item","tableRange":"0..1","shardingColumn":"order_id ","algorithmExpression":"order_id %2"}]
  13. //默认分库规则
  14. vivo.it.sharding.default.db.rule.config = {"shardingColumn":"user_id","algorithmExpression":"user_id %2"}

五、使用建议

结合官方文档和业务实践经验，我们也梳理了部分使用Sharding-JDBC的建议供大家参考，实际具体如何优化SQL写法（比如子查询、分页、分组排序等）还需要结合业务的实际场景来进行测试和调优。

（1）强制等级

  • 建议①：涉及分片表的SQL必须携带分片键

  • 原因：无分片键会导致全路由，存在严重的性能隐患

  • 建议②：禁止一条SQL中的分片值路由至不同的库

  • 原因：跨库操作存在严重的性能隐患，事务操作会升级为分布式事务，增加业务复杂度

  • 建议③：禁止对分片键使用运算表达式或函数操作

  • 原因：无法提前计算表达式和函数获取分片值，导致全路由

  • 说明：详见官方文档

图片

图片

  • 建议⑤：包含CASE WHEN、HAVING、UNION (ALL)语法的分片SQL，不支持路由至多数据节点

  • 说明：详见官方文档

（2）建议等级

  • ② 建议跨多表的分组SQL的分组字段与排序字段保证一致

  • 原因：分组和排序字段不一致只能通过内存合并，大数据量时存在性能隐患

  • 说明：详见官方文档

  • ③ 建议通过全局递增的分布式id来优化分页查询

  • 原因：Sharding-JDBC的分页优化侧重于结果集的流式合并来避免内存爆涨，但深度分页自身的性能问题并不能解决

  • 说明：详见官方文档

六、总结

本文结合个人理解梳理了各个引擎的源码入口和关键逻辑，读者可以结合本文和官方文档更好的定位理解Sharding-JDBC的源码实现。定制开发的目的是为了降低业务接入成本，尽可能减少业务存量SQL的改造，部分改造思想其实与官方社区也存在差异，比如跳过语法解析，官方社区致力于通过优化解析引擎来适配各种语法，而不是跳过解析阶段，可参考官方issue。源码分析和定制改造只涉及了Sharding-JDBC的数据分片和读写分离功能，定制开发的功能也在生产环境经过了考验，如有不足和优化建议，也欢迎大家批评指正。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/410247
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号