shardingjdbc动态修改分表规则

shardingsphere中分为shardingshpere-jdbc、shardingsphere-proxy、shardingshpere-scaling三个项目，当前遇到一个动态生成表结构的问题，每次动态生成一个表的时候。在不重启服务的前提下，关于该表的分表规则就需要改变(如:[1,2]的规则需要改写[1,2,3])。这种场景通过shardingsphere-proxy是可以实现的。可以看这个issue https://github.com/apache/shardingsphere/issues/16725。

鉴于当前项目没有使用注册中心等服务，查看shardingjdbc的ShardingDataSource 源码发现，只有在每次获取Connection的时候才会去获取分表规则，关键点在getConnection()。而在ShardingDataSource中getConnection为final方法，我们无法重写。并且它只提供了DataSource的getConnection实现，深入思考，自己实现一个ShardingDataSource，只要getConnection的时候修改一下runtimeContext为动态的处理方法，就能实现动态的加载分表规则，而且改动范围可控。
demo地址: https://gitee.com/longguiyunjosh/shardingjdbc-dynamic-rules

具体方法如下

*shardingjdbc的ShardingDataSource 源码

@Getter
public class ShardingDataSource extends AbstractDataSourceAdapter {
    
    private final ShardingRuntimeContext runtimeContext;
    
    static {
        NewInstanceServiceLoader.register(RouteDecorator.class);
        NewInstanceServiceLoader.register(SQLRewriteContextDecorator.class);
        NewInstanceServiceLoader.register(ResultProcessEngine.class);
    }
    
    public ShardingDataSource(final Map<String, DataSource> dataSourceMap, final ShardingRule shardingRule, final Properties props) throws SQLException {
        super(dataSourceMap);
        checkDataSourceType(dataSourceMap);
        runtimeContext = new ShardingRuntimeContext(dataSourceMap, shardingRule, props, getDatabaseType());
    }
    
    private void checkDataSourceType(final Map<String, DataSource> dataSourceMap) {
        for (DataSource each : dataSourceMap.values()) {
            Preconditions.checkArgument(!(each instanceof MasterSlaveDataSource), "Initialized data sources can not be master-slave data sources.");
        }
    }
    
/**
*  每次创建一个Connection的时候才会获取runtimeConext(分库分表规则)。
*/
    @Override
    public final ShardingConnection getConnection() {
        return new ShardingConnection(getDataSourceMap(), runtimeContext, TransactionTypeHolder.get());
    }
}

• 参考ShardingDataSource自定义一个DataSource

/**
 * 动态修改分表规则核心类
 *
 * @see ShardingDataSource
 */
@Slf4j
public class DynamicRulesDatasource extends AbstractDataSourceAdapter {

    private AtomicLong version;
    private Map<Long, ShardingRuntimeContext> shardingRuntimeContextMap;

    @Getter
    private Properties properties;

    public DynamicRulesDatasource(Map<String, DataSource> dataSourceMap, ShardingRule shardingRule, Properties props) throws SQLException {
        super(dataSourceMap);
        this.checkDataSourceType(dataSourceMap);
        ShardingRuntimeContext runtimeContext = new ShardingRuntimeContext(dataSourceMap, shardingRule, props, this.getDatabaseType());
        version = new AtomicLong(0);
        properties = props;
        shardingRuntimeContextMap = new ConcurrentHashMap<>();
        shardingRuntimeContextMap.put(version.getAndIncrement(), runtimeContext);

    }

    private void checkDataSourceType(Map<String, DataSource> dataSourceMap) {
        Iterator var2 = dataSourceMap.values().iterator();

        while (var2.hasNext()) {
            DataSource each = (DataSource) var2.next();
            Preconditions.checkArgument(!(each instanceof MasterSlaveDataSource), "Initialized data sources can not be master-slave data sources.");
        }

    }

    public final ShardingConnection getConnection() {

        return new ShardingConnection(this.getDataSourceMap(), getRuntimeContext(), TransactionTypeHolder.get());
    }

     /**
     * 设置分库分表规则
     * @param shardingRule
     * @param props
     * @throws SQLException
     */
    public void setRuntimeContext(ShardingRule shardingRule, Properties props) throws SQLException {
        ShardingRuntimeContext runtimeContext = new ShardingRuntimeContext(this.getDataSourceMap(), shardingRule, props, this.getDatabaseType());
        shardingRuntimeContextMap.put(version.getAndIncrement(),runtimeContext);
        this.resizeMap();
    }

    private void resizeMap(){

        Set<Long> keySet = shardingRuntimeContextMap.keySet();
        if(keySet.size() < 3){
            return;
        }

        List<Long> keys = keySet.stream().collect(Collectors.toList());
        keys.sort(Long::compareTo);

        //预留两个缓存，保证修改分表规则的时候，永远不会获取到null
        for (int i = 0; i < keys.size() - 2; i++) {
            long key = keys.get(i);
            if(!this.shardingRuntimeContextMap.containsKey(key)){
                continue;
            }
            this.shardingRuntimeContextMap.remove(key);
        }
    }

    @Generated
    public ShardingRuntimeContext getRuntimeContext() {
        //永远取最新的分表规则
        return shardingRuntimeContextMap.get(version.get() == 0 ? 0: version.get() - 1);
    }

    static {
        NewInstanceServiceLoader.register(RouteDecorator.class);
        NewInstanceServiceLoader.register(SQLRewriteContextDecorator.class);
        NewInstanceServiceLoader.register(ResultProcessEngine.class);
    }

}

• 参考ShardingDataSourceFactory实现DynamicRulesShardingDataSourceFactory来给自定DynamicRulesDatasource的提供创建工作

/**
 *
 * @see ShardingDataSourceFactory
 */
public class DynamicRulesShardingDataSourceFactory {

    public static DataSource createDataSource(Map<String, DataSource> dataSourceMap, ShardingRuleConfiguration shardingRuleConfig, Properties props) throws SQLException {
        return new DynamicRulesDatasource(dataSourceMap, new ShardingRule(shardingRuleConfig, dataSourceMap.keySet()), props);
    }

    @Generated
    private DynamicRulesShardingDataSourceFactory() {
    }

}

• datasource初始化DynamicRuleDatasourceConfig

@Configuration
public class DynamicRuleDatasourceConfig {

    @Bean
    public DataSource init() throws SQLException {
        // 配置真实数据源
        Map<String, DataSource> dataSourceMap = new HashMap<>();

        // 配置第一个数据源
        HikariDataSource dataSource1 = new HikariDataSource();
        dataSource1.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource1.setJdbcUrl("jdbc:mysql://localhost:3306/ds0");
        dataSource1.setUsername("root");
        dataSource1.setPassword("sinA01234.");
        dataSourceMap.put("ds0", dataSource1);

        Properties properties = new Properties();
        properties.setProperty("sql.show", "true");

        // 获取数据源对象
        DataSource dataSource = DynamicRulesShardingDataSourceFactory.createDataSource(dataSourceMap, initRuleConfig(), properties);
        return dataSource;
    }

    public ShardingRuleConfiguration initRuleConfig(){

        // 配置Order表规则
        TableRuleConfiguration orderTableRuleConfig = new TableRuleConfiguration("t_order","ds0.t_order_${[1,2]}");
        // 配置分库 + 分表策略
        //orderTableRuleConfig.setDatabaseShardingStrategyConfig(new InlineShardingStrategyConfiguration("user_id", "ds${user_id % 2}"));
        orderTableRuleConfig.setTableShardingStrategyConfig(new InlineShardingStrategyConfiguration("tenant_id", "t_order_${tenant_id}"));

        // 配置Order表规则
        TableRuleConfiguration orderItemTableRuleConfig = new TableRuleConfiguration("t_order_item","ds0.t_order_item_${[1,2]}");
        //分表策略
        orderItemTableRuleConfig.setTableShardingStrategyConfig(new InlineShardingStrategyConfiguration("tenant_id", "t_order_item_${tenant_id}"));

        List<String> bindingTableGroups = new ArrayList<>();
        bindingTableGroups.add("t_order");
        bindingTableGroups.add("t_order_item");

        // 配置分片规则
        ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
        shardingRuleConfig.getTableRuleConfigs().add(orderTableRuleConfig);
        shardingRuleConfig.getTableRuleConfigs().add(orderItemTableRuleConfig);
        shardingRuleConfig.setBindingTableGroups(bindingTableGroups);

        return shardingRuleConfig;
    }
}