Sharding-JDBC 源码之 SQL 路由

╰+哭是因爲堅強的太久メ 2022-12-30 15:59 135阅读 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]
6.  第六篇 [Sharding-JDBC 源码之结果集归并][Sharding-JDBC 1]

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

接上文，在完成 SQL 解析之后，也就意味着我们拿到了逻辑 SQL 的抽象语法树及解析结果，  
例如，以下SQL：

select goods_name from t_goods where goods_id=1 and type=0

解析之后的为抽象语法树见下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center]

接下来，我们将继续学习SQL 路由过程。

### 目录 ###

*   *   *  Sharding-JDBC 系列
         *  路由类型
         *   *  根据分片键划分路由
             *  根据分片键场景划分路由
             *  不携带分片键的广播路由
         *  SQL 路由
         *  总结

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

### 路由类型 ###

#### 根据分片键划分路由 ####

根据解析上下文匹配数据库和表的分片策略,生成路由路径。对于携带分片键的SQL，根据分片键的不同可以划分不同的路由：

*  单片路由：分片键的操作符是 **=**；
 *  多片路由：分片键的操作符是 **IN**；
 *  范围路由：分片键的操作符是 **BETWEEN**；
 *  广播路由: 不携带分片键的SQL。

#### 根据分片键场景划分路由 ####

根据分片键进行路由的场景，又细分以下三种类型:

*  直接路由（强制路由）:需使用**HintAPI**直接指定路由至库表，可适用分片键不在SQL中场景；
 *  标准路由（推荐）:适用范围是不包含关联查询或仅包含绑定表之间关联查询的SQL，在分片运算符是 **BETWEEN** 或 **IN** 时，单表查询或关联查询生成的真实SQL数量是一致的；
 *  笛卡尔积路由:无法根据绑定表的关系定位分片规则，非绑定表之间的关联查询需要拆解为笛卡尔积组合执行。

#### 不携带分片键的广播路由 ####

不携带分片键的 SQL，则采取广播路由的方式，根据SQL类型又可以划分为以下5种：

*  全库表路由：处理对数据库中与其逻辑表相关的所有真实表的操作，主要包括不带分片键的 **DQL** 和 **DML**，以及 **DDL** 等；
 *  全库路由：处理数据库相关命令，用于库设置的 **SET** 类型的数据库管理命令，以及 **TCL** 这样的事务控制语句；
 *  全实例路由：用于 **DCL** 操作，授权语句针对的是数据库的实例；例如：`CREATE USER xxx`；
 *  单播路由：获取某一真实表信息的场景，它仅需要从任意库中的任意真实表中获取数据即可。例如：`DESCRIBE t_goods`;
 *  阻断路由：用于屏蔽SQL对数据库的操作，命令不会在真实数据库中执行例如:`USE order_db`。

### SQL 路由 ###

介绍完分片键类型，接下来我们从源码上来看是如何路由。以下继续以 **select** 查询为例，在上篇文章中，我们拿到了 `SQLStatement` 的实例 `SelectStatement` 对象，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]  
在拿到 `SelectStatement` 对象后，开始进行路由，非 **HintSQL** 使用 `ParsingSQLRouter` 进行路由，执行 `ParsingSQLRouter`\#route 方法：

public SQLRouteResult route(final String logicSQL, final List<Object> parameters, final SQLStatement sqlStatement) { 
            GeneratedKey generatedKey = null;
            // 如果是插入语句，获取生成的主键
            if (sqlStatement instanceof InsertStatement) { 
                generatedKey = getGenerateKey(shardingRule, (InsertStatement) sqlStatement, parameters);
            }
            SQLRouteResult result = new SQLRouteResult(sqlStatement, generatedKey);
            // 通过优化引擎获取分库分表条件，存放在 ShardingConditions 对象中
            ShardingConditions shardingConditions = OptimizeEngineFactory.newInstance(shardingRule, sqlStatement, parameters, generatedKey).optimize();
            
            // ......省略，先分析优化引擎
        }

在上面代码片段中，优化引擎会去获取分库（表）的条件映射，看下 `QueryOptimizeEngine#optimize` 方法：

public ShardingConditions optimize() { 
            List<ShardingCondition> result = new ArrayList<>(orCondition.getAndConditions().size());
            for (AndCondition each : orCondition.getAndConditions()) { 
                // 调用重载方法 optimize 封装 ShardingCondition 对象
                // 1. each.getConditionsMap() 按照列进行分组
                result.add(optimize(each.getConditionsMap()));
            }
            return new ShardingConditions(result);
        }
        
        // 重载方法 optimize 封装 ShardingCondition 对象
        private ShardingCondition optimize(final Map<Column, List<Condition>> conditionsMap) { 
            ShardingCondition result = new ShardingCondition();
            // 2. 遍历每个 key，封装 ShardingValue 对象
            for (Entry<Column, List<Condition>> entry : conditionsMap.entrySet()) { 
                // ...... 省略 try..catch
                // 调用重载方法 optimize 进行处理
                ShardingValue shardingValue = optimize(entry.getKey(), entry.getValue());
                if (shardingValue instanceof AlwaysFalseShardingValue) { 
                    return new AlwaysFalseShardingCondition();
                }
                result.getShardingValues().add(shardingValue);
            }
            return result;
        }
        
        // 调用重载方法 optimize 封装 ShardingValue 对象
        private ShardingValue optimize(final Column column, final List<Condition> conditions) { 
            List<Comparable<?>> listValue = null;
            Range<Comparable<?>> rangeValue = null;
            for (Condition each : conditions) { 
                // 3. 分键值必须是可比较对象，即继承 Comparable 接口
                List<Comparable<?>> conditionValues = each.getConditionValues(parameters);
                if (ShardingOperator.EQUAL == each.getOperator() || ShardingOperator.IN == each.getOperator()) { 
                    // 4.在列名相同时，取多个键值的交集，否则返回 AlwaysFalseShardingValue 无法分库（表）
                    listValue = optimize(conditionValues, listValue);
                    if (listValue.isEmpty()) { 
                        return new AlwaysFalseShardingValue();
                    }
                }
                if (ShardingOperator.BETWEEN == each.getOperator()) { 
                    try { 
                        // 在列名相同时，取多个键值最小区间
                        rangeValue = optimize(Range.range(conditionValues.get(0), BoundType.CLOSED, conditionValues.get(1), BoundType.CLOSED), rangeValue);
                    } catch (final IllegalArgumentException ex) { 
                        return new AlwaysFalseShardingValue();
                    }
                }
            }
            // ......省略
        }
        
        private List<Comparable<?>> optimize(final List<Comparable<?>> value1, final List<Comparable<?>> value2) { 
            if (null == value2) { 
                return value1;
            }
            // 取交集
            value1.retainAll(value2);
            return value1;
        }

1.  在调用 `each.getConditionsMap()` 时，会将条件按照列名进行分组，这样相同列的条件就会出现在同一个 Map 对象中，如：

select goods_name from t_goods where goods_id = 1 and goods_id=2 and type=0;

会生成两个 **key**: goods\_id、type，goods\_id 对应的 List 中会存两个对象，为后面的优化合并做准备；  
2. 遍历 Map 对象，获取每个列 key 对应的 `ShardingValue` 对象；  
3. 分键值必须继承 `Comparable` 接口，否则抛出异常；  
4. 这一步是重点，如果一个列 key 出现多个值，则会取他们的交集作为键值。如：上面的 SQL 语句中 goods\_id 出现两次，则取键值 1 和 2 的交集，返回空列表。因为既要查 goods\_id = 1 又要查 goods\_id = 2 是无法完成查询的的，所以进行优化，返回 `AlwaysFalseShardingValue` 对象；  
5. 返回 `AlwaysFalseShardingValue` 对象后，获取到的路由引擎是 `UnicastRoutingEngine`，即单播路由引擎，也就是从真实节点中任意取一个执行该 SQL 语句，此处便不再展开，感兴趣可以去看看 UnicastRoutingEngine\#route 是如何获取路由结果的。

接下来看下正常 SQL 语句的路由详情，继续看 **ParsingSQLRouter\#route** 的中间部分：

public SQLRouteResult route(final String logicSQL, final List<Object> parameters, final SQLStatement sqlStatement) { 
            // ......省略
            
            // 从优化引擎中获取分区条件
            ShardingConditions shardingConditions = OptimizeEngineFactory.newInstance(shardingRule, sqlStatement, parameters, generatedKey).optimize();
            // 根据分区条件及SQL解析结果获取路由引擎
            RoutingResult routingResult = route(parameters, sqlStatement, shardingConditions);
            // ......省略
        }
        
        // 根据条件获取路由引擎
        private RoutingResult route(final List<Object> parameters, final SQLStatement sqlStatement, final ShardingConditions shardingConditions) { 
            Collection<String> tableNames = sqlStatement.getTables().getTableNames();
            RoutingEngine routingEngine;
            if (sqlStatement instanceof UseStatement) { 
                routingEngine = new IgnoreRoutingEngine();
            } else if (sqlStatement instanceof DDLStatement) { 
                routingEngine = new TableBroadcastRoutingEngine(shardingRule, sqlStatement);
            } else if (sqlStatement instanceof ShowDatabasesStatement || sqlStatement instanceof ShowTablesStatement) { 
                routingEngine = new DatabaseBroadcastRoutingEngine(shardingRule);
            } else if (shardingConditions.isAlwaysFalse()) { 
                // 单播路由引擎
                routingEngine = new UnicastRoutingEngine(shardingRule, tableNames);
            } else if (sqlStatement instanceof DALStatement) { 
                routingEngine = new UnicastRoutingEngine(shardingRule, tableNames);
            } else if (tableNames.isEmpty() && sqlStatement instanceof SelectStatement) { 
                routingEngine = new UnicastRoutingEngine(shardingRule, tableNames);
            } else if (tableNames.isEmpty()) { 
                routingEngine = new DatabaseBroadcastRoutingEngine(shardingRule);
            } else if (1 == tableNames.size() || shardingRule.isAllBindingTables(tableNames) || shardingRule.isAllInDefaultDataSource(tableNames)) { 
                // 标准路由引擎
                routingEngine = new StandardRoutingEngine(shardingRule, tableNames.iterator().next(), shardingConditions);
            } else { 
                // TODO config for cartesian set
                routingEngine = new ComplexRoutingEngine(shardingRule, parameters, tableNames, shardingConditions);
            }
            // 根据实际引擎进行路由
            return routingEngine.route();
        }

上面会获取实际的路由引擎，本例中 **tableNames.size() == 1**，所以获得的是 `StandardRoutingEngine`，下面看下该引擎下的 route 方法：

public RoutingResult route() {
            // 根据逻辑表名获得逻辑表分库（表）策略和实际节点
            TableRule tableRule = shardingRule.getTableRule(logicTableName);
            // 获取分库列
            Collection<String> databaseShardingColumns = shardingRule.getDatabaseShardingStrategy(tableRule).getShardingColumns();
            // 获取分表列
            Collection<String> tableShardingColumns = shardingRule.getTableShardingStrategy(tableRule).getShardingColumns();
            Collection<DataNode> routedDataNodes = new LinkedHashSet<>();
            if (HintManagerHolder.isUseShardingHint()) { // 强制分库（表）
                // ......省略
            } else {
                // 1. 分片条件为空时，执行全库表路由，性能较差
                if (shardingConditions.getShardingConditions().isEmpty()) {
                    routedDataNodes.addAll(route(tableRule, Collections.<ShardingValue>emptyList(), Collections.<ShardingValue>emptyList()));
                } else {
                    for (ShardingCondition each : shardingConditions.getShardingConditions()) {
                        // 获取实际分库的键值对映射
                        List<ShardingValue> databaseShardingValues = getShardingValues(databaseShardingColumns, each);
                        // 获取实际分表的键值对映射
                        List<ShardingValue> tableShardingValues = getShardingValues(tableShardingColumns, each);
                        // 2. 获取实际库表节点
                        Collection<DataNode> dataNodes = route(tableRule, databaseShardingValues, tableShardingValues);
                        routedDataNodes.addAll(dataNodes);
                        if (each instanceof InsertShardingCondition) {
                            ((InsertShardingCondition) each).getDataNodes().addAll(dataNodes);
                        }
                    }
                }
            }
            return generateRoutingResult(routedDataNodes);
        }

1.  在 shardingConditions 为空时，表示分库（表）的键值对不存在，无法定位具体库表节点，因此会进行**全库表路由**，例如：

select goods_id from t_goods from goods_name = '商品';

则会遍历所有数据库中的所有表，逐一匹配逻辑表和真实表名，能够匹配得上则执行。路由后成为：

select goods_id from d_sharding_demo1.t_goods_0 from goods_name = '商品';
    select goods_id from d_sharding_demo1.t_goods_1 from goods_name = '商品';
    select goods_id from d_sharding_demo2.t_goods_0 from goods_name = '商品';
    select goods_id from d_sharding_demo2.t_goods_1 from goods_name = '商品';

1.  从 `shardingConditions` 中获取分库（表）的键值对存在时，则执行 `route` 方法获取实际库表节点。

接下来看如何获得实际的数据库节点的，首先是拿到数据库实际节点：

// 获取实际节点
        private Collection<DataNode> route(final TableRule tableRule, final List<ShardingValue> databaseShardingValues, final List<ShardingValue> tableShardingValues) {
            Collection<String> routedDataSources = routeDataSources(tableRule, databaseShardingValues);
            // ...... 省略
        }
        
        // 获取实际数据库节点
        private Collection<String> routeDataSources(final TableRule tableRule, final List<ShardingValue> databaseShardingValues) {
            // 1. 获取实际数据源名称列表
            Collection<String> availableTargetDatabases = tableRule.getActualDatasourceNames();
            if (databaseShardingValues.isEmpty()) {
                return availableTargetDatabases;
            }
            // 2. 根据具体的分片算法获取数据源名称
            Collection<String> result = new LinkedHashSet<>(shardingRule.getDatabaseShardingStrategy(tableRule).doSharding(availableTargetDatabases, databaseShardingValues));
            Preconditions.checkState(!result.isEmpty(), "no database route info");
            return result;
        }
        
        public Collection<String> doSharding(final Collection<String> availableTargetNames, final Collection<ShardingValue> shardingValues) {
            ShardingValue shardingValue = shardingValues.iterator().next();
            // 3. 根据不同类型执行不同的分片算法
            Collection<String> shardingResult = shardingValue instanceof ListShardingValue
                    ? doSharding(availableTargetNames, (ListShardingValue) shardingValue) : doSharding(availableTargetNames, (RangeShardingValue) shardingValue);
            Collection<String> result = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
            result.addAll(shardingResult);
            return result;
        }

1.  首先从逻辑表策略中拿到全部数据源名称列表；
2.  从逻辑表获取分库策略，并执行 `doSharding` 方法，获取实际数据源名称；
3.  根据分片键值类型执行不同的算法，当操作符是**等于**或 **IN** 时，也就是 `ListShardingValue` 类型时会调用 `preciseShardingAlgorithm` 中的 `doSharding` 方法进行精确分片；当操作符是 `BETWEEN` 时，也就是 `RangeShardingValue` 类型时，执行 `rangeShardingAlgorithm` 的 `doSharding` 方法进行范围分片，而这两个方法都是可以自定义实现并在配置文件中配置好；
4.  执行完 `doSharding` 方法后，也就意味着我们拿到了实际待执行的数据源名称，下一步是找到该数据源下的实际表节点。

回到拿到数据源那一步：

private Collection<DataNode> route(final TableRule tableRule, final List<ShardingValue> databaseShardingValues, final List<ShardingValue> tableShardingValues) {
            // 已拿到实际数据源名称
            Collection<String> routedDataSources = routeDataSources(tableRule, databaseShardingValues);
            // 遍历数据源，获取实际表节点
            for (String each : routedDataSources) {
                // 根据数据源及分表策略获取实际表节点
                result.addAll(routeTables(tableRule, each, tableShardingValues));
            }
            return result;
        }
        
        private Collection<DataNode> routeTables(final TableRule tableRule, final String routedDataSource, final List<ShardingValue> tableShardingValues) {
            // 1. 获取数据源下全部实际表节点
            Collection<String> availableTargetTables = tableRule.getActualTableNames(routedDataSource);
            // 2. 获取分表策略，执行 doSharding
            Collection<String> routedTables = new LinkedHashSet<>(tableShardingValues.isEmpty() ? availableTargetTables
                    : shardingRule.getTableShardingStrategy(tableRule).doSharding(availableTargetTables, tableShardingValues));
            Preconditions.checkState(!routedTables.isEmpty(), "no table route info");
            Collection<DataNode> result = new LinkedList<>();
            for (String each : routedTables) {
                result.add(new DataNode(routedDataSource, each));
            }
            return result;
        }
        
        private Collection<String> doSharding(final ListShardingValue shardingValue) {
            Collection<String> result = new LinkedList<>();
            for (PreciseShardingValue<?> each : transferToPreciseShardingValues(shardingValue)) {
                // 3. execute 会执行 groovy 表达式
                result.add(execute(each));
            }
            return result;
        }
        
        private String execute(final PreciseShardingValue shardingValue) {
            Closure<?> result = closure.rehydrate(new Expando(), null, null);
            result.setResolveStrategy(Closure.DELEGATE_ONLY);
            result.setProperty(shardingValue.getColumnName(), shardingValue.getValue());
            // 4. 执行 groovy 脚本
            return result.call().toString();
        }

1.  获取数据源下全部实际表节点；
2.  获取配置文件中配置的分表策略，本例中使用的是 `InlineShardingStrategy`，因此会执行该策略下 `doSharding` 实现方法；
3.  在 `doSharding` 中会调用 `execute` 方法；
4.  `execute` 会根据分表键值执行 **groovy** 脚本，计算出实际表节点。

经过上面步骤，已获取实际数据源名称和实际表节点，将结果封装为 `RoutingResult` 对象并返回

private RoutingResult generateRoutingResult(final Collection<DataNode> routedDataNodes) { 
            RoutingResult result = new RoutingResult();
            for (DataNode each : routedDataNodes) { 
                // 数据源名称
                TableUnit tableUnit = new TableUnit(each.getDataSourceName());
                // RoutingTable 对象封装逻辑表与实际表之间的映射
                tableUnit.getRoutingTables().add(new RoutingTable(logicTableName, each.getTableName()));
                // 封装为 RoutingResult 对象
                result.getTableUnits().getTableUnits().add(tableUnit);
            }
            return result;
        }

到这里，SQL 路由就全部完成了，接下来将会根据路由结果进行SQL改写，下一章继续分析。

### 总结 ###

1.  首先介绍了路由的类型，根据分片键类型和使用场景划分类型，以及不带分片键的广播路由；
2.  接下来通过 **DQL** 语句 **select** 分析了路由的源码；
    
    1.  首先会从优化引擎 `QueryOptimizeEngine` 中获取分片键值对映射列表，并会在获取的同时进行路由条件优化，避免不必要的分库分表；
    2.  接着根据分区条件 `ShardingConditions` 对象 及SQL解析结果 `SelectStatement` 对象获取路由引擎，本例获取的是 `StandardRoutingEngine` 标准路由引擎；
    3.  在进行路由时，如果 `ShardingConditions` 对象为空，则会进行全库表路由，此时会将全部数据源的逻辑表对应的真实表全部查询一遍，性能较差，需要格外注意；
    4.  `ShardingConditions` 不为空，则按照逻辑表配置的分库（表）策略算法，执行各自 `doSharding` 方法获取实际的数据源名称和实际表节点；
    5.  将路由得到的数据源名称和表节点封装在 `RoutingResult` 对象中返回。
3.  拿到 `RoutingResult` 对象后，为我们下一步进行 SQL 改写做好充分准备。

[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
[Sharding-JDBC 1]: https://blog.csdn.net/smile_from_2015/article/details/112101676
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/e4b8c318ab4b463d8d629d1d6fa220ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NtaWxlX2Zyb21fMjAxNQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/cc185da4271d4f2a953e6dbed54a30d5.png