shardingsphere运行期创建表且分表_shardingsphere自动建表

使用背景

最近一个项目上使用了shardingsphere来做分库。常规使用是先在数据库创建表，然后配置到shardingsphere实现分表；然而本项目是根据业务操作后生成的表，需要在运行期进行动态分表。网上找的动态分表可能是版本问题还是有很多要改的地方，为方便后续使用特此记录一下。

流程

1. shardingsphere只配置数据源，不配置分表规则
2. 根据业务生成表
3. 将表装载到shardingsphere，后续查询才能分表查询
4. 项目启动时将需要分表的表装载到shardingsphere

文章主要针对第三步

开始使用

maven依赖

	<dependency>
         <groupId>org.apache.shardingsphere</groupId>
         <artifactId>sharding-jdbc-spring-boot-starter</artifactId>
         <version>4.1.1</version>
     </dependency>
1
2
3
4
5

application

spring.shardingsphere.datasource.names=m1
spring.shardingsphere.datasource.m1.type=com.alibaba.druid.pool.DruidDataSource
spring.shardingsphere.datasource.m1.driver-class-name=com.mysql.cj.jdbc.Driver
spring.shardingsphere.datasource.m1.url=jdbc:mysql://xxx:3306/xxx?useSSL=false&characterEncoding=utf8&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true&&rewriteBatchedStatements=true&allowMultiQueries=true
spring.shardingsphere.datasource.m1.username=xxx
spring.shardingsphere.datasource.m1.password=xxx

spring.shardingsphere.props.sql.show=true
spring.main.allow-bean-definition-overriding=true
1
2
3
4
5
6
7
8
9

分表配置

我的分表是根据外键id进行取模0和1，表名为：xxx_xxx_0 和 xxx_xxx_1，不同规则要做修改

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.api.config.sharding.TableRuleConfiguration;
import org.apache.shardingsphere.api.config.sharding.strategy.InlineShardingStrategyConfiguration;
import org.apache.shardingsphere.core.rule.ShardingRule;
import org.apache.shardingsphere.core.rule.TableRule;
import org.apache.shardingsphere.core.strategy.keygen.SnowflakeShardingKeyGenerator;
import org.apache.shardingsphere.core.strategy.route.inline.InlineShardingStrategy;
import org.apache.shardingsphere.shardingjdbc.jdbc.core.datasource.ShardingDataSource;
import org.apache.shardingsphere.sql.parser.binder.metadata.table.TableMetaData;
import org.apache.shardingsphere.sql.parser.binder.metadata.table.TableMetaDataLoader;
import org.apache.shardingsphere.underlying.common.metadata.ShardingSphereMetaData;
import org.apache.shardingsphere.underlying.common.rule.DataNode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.sql.DataSource;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Component
public class ShardingTableRuleActualDataNodesRefresh {


    @Autowired
    private DataSource dataSource;


    public void refreshActualDataNodes(String dynamicTableName) {
        log.info("Job 动态刷新 actualDataNodes START");
        ShardingDataSource shardingDataSource = (ShardingDataSource) dataSource;
        TableRule tableRule = null;
        try {
            ShardingSphereMetaData metaData = shardingDataSource.getRuntimeContext().getMetaData();
            TableMetaData tableMetaData = TableMetaDataLoader.load(dataSource, dynamicTableName + "_0", shardingDataSource.getDatabaseType().getName());
            metaData.getSchema().put(dynamicTableName, tableMetaData);
            ShardingRule shardingRule = shardingDataSource.getRuntimeContext().getRule();
            tableRule = shardingRule.getTableRule(dynamicTableName);
        } catch (Exception e) {
            log.error(String.format("逻辑表：%s 动态分表配置错误！", dynamicTableName));
        }
        String dataSourceName = tableRule.getActualDataNodes().get(0).getDataSourceName();
        String logicTableName = tableRule.getLogicTable();
        assert tableRule != null;
        List<DataNode> newDataNodes = getDataNodes(dataSourceName, logicTableName);
        if (newDataNodes.isEmpty()) {
            throw new UnsupportedOperationException();
        }

        try {
            dynamicRefreshDatasource(dataSourceName, dynamicTableName, tableRule, newDataNodes);
        } catch (Exception e) {
            e.printStackTrace();
        }

        TableRuleConfiguration configuration = new TableRuleConfiguration(dynamicTableName, dataSourceName + "." + dynamicTableName + "_${0..1}");
        InlineShardingStrategyConfiguration a = new InlineShardingStrategyConfiguration("monitor_point_id", dynamicTableName + DBUtil.shardingStrategy());
        configuration.setTableShardingStrategyConfig(a);
        Collection<TableRuleConfiguration> ruleConfigs = shardingDataSource.getRuntimeContext().getRule().getRuleConfiguration().getTableRuleConfigs();
        ruleConfigs.add(configuration);

        log.info("Job 动态刷新 actualDataNodes END");
    }


    /**
     * 获取数据节点
     */
    private List<DataNode> getDataNodes(String dataSourceName, String logicTableName) {
        Set<DataNode> newDataNodes = Sets.newHashSet();
        newDataNodes.add(new DataNode(dataSourceName + "." + logicTableName + "_0"));
        newDataNodes.add(new DataNode(dataSourceName + "." + logicTableName + "_1"));
        // 扩展点
        return Lists.newLinkedList(newDataNodes);
    }

    /**
     * 动态刷新数据源
     */
    private void dynamicRefreshDatasource(String dataSourceName, String dynamicTableName, TableRule tableRule, List<DataNode> newDataNodes)
            throws NoSuchFieldException, IllegalAccessException {
        Set<String> actualTables = Sets.newHashSet();
        Map<DataNode, Integer> dataNodeIndexMap = Maps.newHashMap();
        AtomicInteger index = new AtomicInteger(0);
        newDataNodes.forEach(dataNode -> {
            actualTables.add(dataNode.getTableName());
            if (index.intValue() == 0) {
                dataNodeIndexMap.put(dataNode, 0);
            } else {
                dataNodeIndexMap.put(dataNode, index.intValue());
            }
            index.incrementAndGet();
        });

        Field generateKeyColumn = TableRule.class.getDeclaredField("generateKeyColumn");
        generateKeyColumn.setAccessible(true);
        generateKeyColumn.set(tableRule, "id");

        Field shardingKeyGenerator = TableRule.class.getDeclaredField("shardingKeyGenerator");
        shardingKeyGenerator.setAccessible(true);
        shardingKeyGenerator.set(tableRule, new SnowflakeShardingKeyGenerator());

        // 动态刷新：actualDataNodesField
        Field actualDataNodesField = TableRule.class.getDeclaredField("actualDataNodes");
        Field modifiersField = Field.class.getDeclaredField("modifiers");
        modifiersField.setAccessible(true);
        modifiersField.setInt(actualDataNodesField, actualDataNodesField.getModifiers() & ~Modifier.FINAL);
        actualDataNodesField.setAccessible(true);
        actualDataNodesField.set(tableRule, newDataNodes);
        // 动态刷新：actualTablesField
        Field actualTablesField = TableRule.class.getDeclaredField("actualTables");
        actualTablesField.setAccessible(true);
        actualTablesField.set(tableRule, actualTables);
        // 动态刷新：dataNodeIndexMapField
        Field dataNodeIndexMapField = TableRule.class.getDeclaredField("dataNodeIndexMap");
        dataNodeIndexMapField.setAccessible(true);
        dataNodeIndexMapField.set(tableRule, dataNodeIndexMap);
        // 动态刷新：datasourceToTablesMapField
        Map<String, Collection<String>> datasourceToTablesMap = Maps.newHashMap();
        datasourceToTablesMap.put(dataSourceName, actualTables);
        Field datasourceToTablesMapField = TableRule.class.getDeclaredField("datasourceToTablesMap");
        datasourceToTablesMapField.setAccessible(true);
        datasourceToTablesMapField.set(tableRule, datasourceToTablesMap);

        Field tableShardingStrategy = TableRule.class.getDeclaredField("tableShardingStrategy");
        tableShardingStrategy.setAccessible(true);
        tableShardingStrategy.set(tableRule, getsStrategy(dynamicTableName));

        ShardingDataSource shardingDataSource = (ShardingDataSource) dataSource;
        ShardingRule shardingRule = shardingDataSource.getRuntimeContext().getRule();
        Collection<TableRule> tableRules = shardingRule.getTableRules();
        tableRules.add(tableRule);
    }

    public InlineShardingStrategy getsStrategy(String tableName) {
        return new InlineShardingStrategy(new InlineShardingStrategyConfiguration("外键ID", tableName + "_${外键ID % 2}"));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

项目启动装载

表是动态创建的，在项目重启后分表规则失效，需要重新装载到shardingsphere

	ShardingSphereMetaData metaData = shardingDataSource.getRuntimeContext().getMetaData();
 	SchemaMetaData schema = metaData.getSchema();
    Collection<String> tableNames = schema.getAllTableNames();
    Set<String> tables = new HashSet<>();
    for (String tableName : tableNames) {
        if (tableName.startsWith("xxx_")) {
            tables.add(tableName.substring(0, tableName.lastIndexOf("_")));
        }
    }
     tables.forEach(name -> {
     	shardingTableRuleActualDataNodesRefresh.refreshActualDataNodes(name);
     });
1
2
3
4
5
6
7
8
9
10
11
12

shardingTableRuleActualDataNodesRefresh.refreshActualDataNodes(name) 根据表名分表。

当时项目较急，没做什么优化。欢迎大佬指正！