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;
}
}
