ShardingSphere笔记（三）：自定义分片算法 — 按月分表·真·自动建表
2024-05-13 05:11:47
ShardingSphere笔记（二）：自定义分片算法 — 按月分表·真·自动建表
文章目录
- ShardingSphere笔记（二）：自定义分片算法 — 按月分表·真·自动建表
- 一、 前言
- 二、 Springboot 的动态数据库
- 三、 实现我们自己的动态数据库
- 1. POM
- 2. 实现 AbstractRoutingDatasource
- 3. 创建数据源
- 四、实现按月分表算法
- 1. 分表算法
- 2. 添加SPI
- 五、自动建表
- 六、 拓展
- 七、源代码
一、 前言
上一篇文章中，已经介绍了shardingshpere 怎么进行按月分表。但是我们知道，一般来说按月分表肯定是本月到了之后才会进行建表，因为我们不可能预想到系统到底可以运行多久。而且在第一篇文章里面也已经提到了，Shardingshpere 默认是要求配置的实际表都是存在的。
难不成我们要每年建一次当年的按月分表的数据库表，每年改一次软件的配置文件然后重启软件吗？且不说如果维护项目的大兄弟离职，锅会留给下一个接手项目的人。这种每年修改一次，提心吊胆的记着改数据库，改软件的苦差事，一点都不优雅。
那么有没有好一点的方法能够让它自动建表呢？欸，这就得说说我在研究的过程中，碰巧碰到了一种骚操作，可以在运行过程中建表，并且能让sharesphere 框架知道新增了表，自动更新表缓存。
因此这篇文章最后能实现下面的这几个效果：
- 使用Shardingsphere 实现按月分表的查询功能。
- 根据查询的语句判断该月的表是否存在，不存在自动建表，并自动刷新到shardingsphere缓存表中。
- 实现多数据库切换功能。并且提供原生的数据库操作数据源，因为从第一篇文章我们已经知道，shardingsphere数据源不支持数据库函数，不支持原生ddl操作，提供原生数据源可以保证我们在原生数据源中执行一些Shardingsphere不支持的操作。
不过、因为是骚操作，可能的缺点有两个：
- 该骚操作可能会受ShardingSphere版本影响，万一哪天ShardingSphere程序员爸爸修改了表缓存逻辑，这一块儿逻辑就有可能失效了（目前版本 5.2.1）
- 数据库里面必须存在一个逻辑表，因为自动建表语句是利用的mysql的 CREATE TABLE LIKE 创建的，你得保证有这么个表我才能建表罢，所以哪怕这个表实际上用不到，数据库中也要存在一个这样的表。
二、 Springboot 的动态数据库
因为要实现多数据源的动态切库，这里需要先介绍Springboot 是怎么做动态数据库切库的。
如果已经有大佬对springboot 的动态数据门儿清的话，可以跳过这一节。
众所周知，好吧，至少刚入职的时候我不知道，Springboot 通过 AbstractRoutingDatasource 切换不同的数据源。默认的springboot是单数据库的数据源。如果我们要实现多数据源的动态切换就需要实现 AbstractRoutingDatasource。 并将该DataSource 作为一个Bean注册到Spring中。
那么 AbstractRoutingDatasource 是怎么做到多数据源的呢？众所~~，好吧不废话了，我们首先看看DataSource的作用是什么。下面是Datasource的源码。
public interface DataSource extends CommonDataSource, Wrapper {/*** <p>Attempts to establish a connection with the data source that* this {@code DataSource} object represents.** @return a connection to the data source* @exception SQLException if a database access error occurs* @throws java.sql.SQLTimeoutException when the driver has determined that the* timeout value specified by the {@code setLoginTimeout} method* has been exceeded and has at least tried to cancel the* current database connection attempt*/Connection getConnection() throws SQLException;
/// ..... 无关紧要的代码
}
简单说呢，就是获取当前数据库的一个Connection，诶嘿，人家可没说这个就只能返回一个数据库的连接，说的只是当前数据库的连接。没错，AbstractRoutingDatasource 作者肯定就是Get到了这个语言上的小差别。
AbstractRoutingDatasource 重写了这个 getConnection.
public abstract class AbstractRoutingDataSource extends AbstractDataSource implements InitializingBean {/*** 配置多个数据源， * key：string唯一标识数据源，* value: datasource 数据源 */@Nullableprivate Map<Object, DataSource> resolvedDataSources;public Connection getConnection() throws SQLException {return this.determineTargetDataSource().getConnection();}/*** 确定当前的数据源*/protected DataSource determineTargetDataSource() {// 获取数据库Id（key）Object lookupKey = this.determineCurrentLookupKey();// 根据Id获取数据源DataSource dataSource = (DataSource)this.resolvedDataSources.get(lookupKey);// 忽略一些校验,返回数据源return dataSource； }/*** 获取当前数据库的Id(Key)* 这个方法就是Springboot留给我们扩展的点了。* 所以呀：* 如果你想实现自己的动态切换数据库的框架，只需要扩展两点就可以了* 1. 拿到你的多个数据源和他们的Id（自己随便指定）组合成一个map 设置给 resolvedDataSources* 2. 实现该方法，确定什么时候用哪一个数据库。*/@Nullableprotected abstract Object determineCurrentLookupKey();
}
三、 实现我们自己的动态数据库
1. POM
在这之前，先把POM文件依赖贴一下吧，否则缺少某个库的函数导致代码复现不出来大家该说耍流氓了。
<!-- 没什么用，辅助Idea做配置文件提示的 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><!-- Lombok 开发神器，懂得都懂 --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><!-- ShardingSphere 核心依赖 --><dependency><groupId>org.apache.shardingsphere</groupId><artifactId>shardingsphere-jdbc-core</artifactId><version>${shardingsphere-jdbc.version}</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><!-- 工具类库，有点大，看需求引入，只是用习惯了 --><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.11</version></dependency><!-- 阿里的 Transmittable ThreadLocal 解决父子线程值传递问题 --><dependency><groupId>com.alibaba</groupId><artifactId>transmittable-thread-local</artifactId><version>2.14.2</version></dependency>
2. 实现 AbstractRoutingDatasource
ShardingSphereDynamicDataSource
public class ShardingSphereDynamicDataSource extends AbstractRoutingDataSource {/*** 多数据库 Map，* 这里重新创建了一个Map对象目的有两个* 1. ards 的 resolvedDataSources 是 private 的，子类无法访问。* 2. 设置为static方式的，可以通过静态方法访问，更方便。*/static final Map<Object, Object> DATA_SOURCE_MAP = new ConcurrentHashMap<>(10);private String defaultDataSource;public ShardingSphereDynamicDataSource(LinkedHashMap<String, DataSource> dataSourceMap) {// 将多数据库Map设置给 resolvedDataSourcessetTargetDataSources(DATA_SOURCE_MAP);dataSourceMap.forEach((dsName, ds) -> {DATA_SOURCE_MAP.put(dsName, ds);if (defaultDataSource == null) {// 设置默认数据源defaultDataSource = dsName;setDefaultTargetDataSource(ds);}});}@Overrideprotected Object determineCurrentLookupKey() {// 通过ThreadLocal 获取数据库Id，防止多线程设置当前数据源导致的并发问题return DynamicDataSourceHelper.dataSourceKey();}/*** 动态添加数据源*/public void addDataSource(String dataSourceKey, DataSource dataSource) {DATA_SOURCE_MAP.put(dataSourceKey, dataSource);afterPropertiesSet();}/*** 获取默认数据源*/public String getDefaultDataSource() {return defaultDataSource;}/*** 根据数据源Id获取数据源*/public DataSource getDataSourceById(String id) {return ((DataSource) DATA_SOURCE_MAP.get(id));}
}
上面的代码比较简单，只要理解了Springboot是怎么做多数据源的，上面代码就很容易看懂了。
不过里面有一个 DynamicDataSourceHelper ，这个是做什么用的呢？
你想啊，我们有了多数据源了，我们是不是要决定什么时候用哪一个数据库？
这个类就是做这个用的，它其实核心就是维护了一个ThreadLocal, 要设置用哪一个数据库的时候，往这个ThreadLocal 里面设置当前数据库的Key， 上面在获取当前数据库的时候就从这个 ThreadLoca里面取值。
为什么要用ThreadLocal 维护呢，直接设置String不行吗？当然不行，多线程中用数据源的时候，到底谁说了算呢？当然最好的方式是每个线程管自己的。用ThreadLocal专业对口嘛。
另外这里用了阿里的 TransmittableThreadLocal， 目的是为了解决父子线程的传值问题。
DynamicDataSourceHelper
public class DynamicDataSourceHelper {/*** 用来判断 ShardingSphere 和 普通数据源的，这里可以先不管它，后面会提到。*/public static final String RAW_DATASOURCE_PREFIX = "sharding::raw::";/*** 使用Alibaba的 TransmittableThreadLocal 解决数据源切换的父子线程之间数据传递的问题*/private static final ThreadLocal<String> DATA_SOURCE_KEY = new TransmittableThreadLocal<>();/*** 设置当前使用的数据源Id*/public static void setDataSource(String dataSourceKey) {DATA_SOURCE_KEY.set(dataSourceKey);}/*** 设置当前使用的原始数据源Id(使用该方法标识使用的不是ShardingSphere数据源，而是原始数据源)*/public static void setRawJdbcDataSource(String dataSourceKey) {if (isRawJdbcDataSourceKey(dataSourceKey)) {DATA_SOURCE_KEY.set(dataSourceKey);} else {DATA_SOURCE_KEY.set(RAW_DATASOURCE_PREFIX + dataSourceKey);}}/*** 使用默认数据源*/public static void defaultDataSource() {DATA_SOURCE_KEY.remove();}/*** 获取当前使用的数据源Id*/public static String dataSourceKey() {return DATA_SOURCE_KEY.get();}/*** 判断该数据源Id是否存在*/public static boolean containsDataSource(String datasource) {return ShardingSphereDynamicDataSource.DATA_SOURCE_MAP.containsKey(datasource);}/*** 根据ShardingShpere 的数据源Id获取到对应的原始数据源Id* 目的是可以获取到ShardingSphere对应的原始数据源。* 还记得上面说我们要能提供原始ddlsql的执行能力吗？* ShardingSphere不支持，单我们可以通过对应的原始数据源来执行。*/public static String rawJdbcDataSourceKey(String key) {if (key.startsWith(RAW_DATASOURCE_PREFIX)) {return key;}return RAW_DATASOURCE_PREFIX + key;}/*** 判断是否是原始数据源，这里默认认为原始数据源的Id比ShardingSphere 的数据源Id多一个前缀* 也就是上面那个静态变量（sharding::raw::）*/ public static boolean isRawJdbcDataSourceKey(String key) {if (StrUtil.isEmpty(key)) {return false;}return key.startsWith(RAW_DATASOURCE_PREFIX);}
}
上面的代码中，已经把怎么切换数据库的逻辑部分写完了，当然它还只是一个架子，还没有具体设置数据源呢，相当于把API给写完了。那这个Api怎么用呢？
public void switchDataSource() {// 切换到 ds1 的ShardingSphere数据源DynamicDataSourceHelper.setDataSource("ds1");mapper.selectSomething();jpaRepo.selectByXXX();// 切换到 ds1 的原始Jdbc数据源DynamicDataSourceHelper.setRawJdbcDataSource("ds1");mapper.selectSomething();jpaRepo.selectByXXX();// 切换到 ds2 的ShardingSphere数据源DynamicDataSourceHelper.setDataSource("ds2");mapper.selectSomething();jpaRepo.selectByXXX();// 切换到 ds2 的原始Jdbc数据源DynamicDataSourceHelper.setRawJdbcDataSource("ds2");mapper.selectSomething();jpaRepo.selectByXXX();
}3. 创建数据源
上面相当于已经把好看的皮囊写完了，灵魂还没有呢，俗话说得好，“好看的皮囊千篇一律，有趣的灵魂万里挑一”， 没有有趣的灵魂怎么能行呢？
这里主要要确定的一个问题是？我们怎么拿到多个数据源？熟悉那些多数据源开源框架的朋友应该比较了解，配置文件嘛。 所以这里大致制定一个规范，如何读取数据源。配置文件结构如下：
spring:shardingsphere:props:# 是否显示 ShardingSpher 的sql，用于Debugsql-show: truedatasource:# 数据源名称列表names: ds1,ds2# 数据源1ds1:type: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.jdbc.Driverjdbc-url: jdbc:mysql://127.0.0.1:3306/test1?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowMultiQueries=trueusername: rootpassword: 123456# 数据2ds2:type: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.jdbc.Driverjdbc-url: jdbc:mysql://127.0.0.1:3306/test2?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowMultiQueries=trueusername: rootpassword: 123456rules:sharding:tables:# 按月分表的数据表（只需要配置逻辑表即可，程序会自动探测存在的表)# 但必须保证逻辑表必须存在，即使它在真实程序中不被使用，因为要通过它自动建表# 这里可能细心的朋友已经发现了，这样写逻辑岂不是所有数据源的分表都一样了吗？# 这里确实，因为公司项目里面就是所有数据源的数据库都是一样的，所以就这样写了。# 如果大家自己的项目里面多个数据源每个数据库表不一样，分表的数据表也不一样，可以改这一块逻辑。每个数据库的分表单独设置。 binding-tables: data, test_data
上面的哦配置文件应该也比较清晰了，这里就不具体解释了，下面就要通过解析该配置文件创建数据库了。
这里就要做一个小约定了，上面我们配置了两个数据源，但是实际上我们要创建两类，四个数据源。它们分别是：
- ShardingSphere 的数据源 ds1, ds2。
- 原始数据源 sharding::raw::ds1, sharding::raw::ds2
还记的上面的原始数据源前缀吗？就是用在这里的，用它加上数据源的Id创建原始数据源的Id。
/*** @Configuration 标识这是一个配置类，Springboot会自动扫描（也方便我们后面制作Springboot-starter）* * @AutoConfigureBefore(DataSourceAutoConfiguration.class) * 我们要保证该配置在Springboot的Datasource配置类之前执行，因为我们要用自己 DIY 的数据源了。得插个队。* @author wangp*/
@Slf4j
@Configuration
@AutoConfigureBefore(DataSourceAutoConfiguration.class)
public class AkxyShardingSphereAutoConfiguration {/*** 读取到有那些表需要进行分表*/@Value("${spring.shardingsphere.rules.sharding.tables.binding-tables}")private List<String> bindingTables;@Beanpublic DataSource dataSource(Environment environment) throws SQLException {Binder binder = Binder.get(environment);Properties properties = binder.bind("spring.shardingsphere.props", Properties.class).orElse(new Properties());// 从配置文件中读取数据源Map<String, DataSource> dataSourceMap = DatasourceHelper.createDataSourceMap(environment);LinkedHashMap<String, DataSource> resultDatasourceMap = new LinkedHashMap<>();for (String dataSourceName : dataSourceMap.keySet()) {DataSource dataSource = dataSourceMap.get(dataSourceName);// 创建ShardingSphere 数据源DataSource shardingSphereDatasource = ShardingSphereDataSourceFactory.createDataSource(dataSource,Collections.singleton(createShardingRuleConfiguration(dataSource)), properties);resultDatasourceMap.put(dataSourceName, shardingSphereDatasource);}// 同时创建一份 原始的 JDBC DataSource， 并且给它们设置前缀，用于区分不同类型的数据源for (String dataSourceName : dataSourceMap.keySet()) {resultDatasourceMap.put(DynamicDataSourceHelper.rawJdbcDataSourceKey(dataSourceName),dataSourceMap.get(dataSourceName));}// 创建动态数据源return new ShardingSphereDynamicDataSource(resultDatasourceMap);}/*** 切片规则配置*/private ShardingRuleConfiguration createShardingRuleConfiguration(DataSource dataSource) throws SQLException {ShardingRuleConfiguration result = new ShardingRuleConfiguration();for (String bindingTable : bindingTables) {// 为每一个分表的数据表创建分表规则result.getTables().add(getTableRuleConfiguration(dataSource, bindingTable));}// 设置分表策略， key: acquisition_time 标识分表的数据列，这里的时间列也是写死的，需要用的朋友注意了，可以写在配置文件里，也可以在代码里面改。// 设置分表算法： his-data-spi-based 为算法Id result.setDefaultTableShardingStrategy(new StandardShardingStrategyConfiguration("acquisition_time", "his-data-spi-based"));// 设置分表算法Map， 上面的算法Id会从这个Map里面找对应的算法// HIS_DATA_SPI_BASED 这个Id是通过 SPI的方式注册的分表算法，SPI的方式具体请参考第二篇文章。result.getShardingAlgorithms().put("his-data-spi-based",new AlgorithmConfiguration("HIS_DATA_SPI_BASED", new Properties()));return result;}/*** 创建分表规则（从ShardingSphere Starter 代码里面抄的~_~）*/private ShardingTableRuleConfiguration getTableRuleConfiguration(DataSource dataSource, String logicTableName) throws SQLException {Set<String> existsActuallyTableSet;try (Connection connection = dataSource.getConnection()) {// 查询实际上存在的表（标规则 logictable_yyyyMM）// 因为这里还没有创建ShardingSphere的数据源呢，这里的数据源还是原始的数据源,还没有被代理，所以可以执行一些原始SQL语句existsActuallyTableSet = MonthShardingDatabaseUtil.queryActuallyTables(connection, logicTableName);// 如果真实表不存在的话, 根据逻辑表创建一个新的表if (CollectionUtil.isEmpty(existsActuallyTableSet)) {// 创建一个当前时间对应的月的表String createTable = logicTableName + MonthShardingDatabaseUtil.shardingSuffix(new Date());if (MonthShardingDatabaseUtil.createTableLike(connection, logicTableName, createTable)) {existsActuallyTableSet = new LinkedHashSet<>();existsActuallyTableSet.add(createTable);log.info("actually table for {} not exists, auto create one : {}", logicTableName, createTable);} else {throw new IllegalStateException("Could not create table " + createTable);}}}// 组合实际存在的表节点列表， 比如： logic_db.data_202301,logic_db_data_202302String actuallyTableNodes = existsActuallyTableSet.stream().map(tableName -> StrUtil.format("{}.{}", "logic_db", tableName)).collect(Collectors.joining(","));ShardingTableRuleConfiguration result = new ShardingTableRuleConfiguration(logicTableName, actuallyTableNodes);// 设置分表策略， key: acquisition_time 标识分表的数据列，这里的时间列也是写死的，需要用的朋友注意了，可以写在配置文件里，也可以在代码里面改。result.setTableShardingStrategy(new StandardShardingStrategyConfiguration("acquisition_time", "his-data-spi-based"));return result;}@Beanpublic AutoCreateTableHelper tableHelper(DataSource dataSource) {// 自动建表Helper，后面会提及return new AutoCreateTableHelper(dataSource);}}上面的代码中，从配置文件里面读取数据源列表，读取分表的数据表列表，然后通过他们来构建ShardingSphere数据源和原始JDBC数据源，并通过他们生成我们上面的动态数据源，注册为Springboot的bean。
上面有两个工具类，DatasourceHelper， MonthShardingDatabaseUtil 这两个工具类一个作用是从配置文件读取Datasource，一个是按月分表的工具类，提供了一个辅助方法。希望别嫌代码多，我也贴出来了，我会把代码也传上去，嫌多的伙伴可以不看这一部分,直接下载源码用。
DatasourceHelper
/*** @author wangp*/
public class DatasourceHelper {private static final String PREFIX = "spring.shardingsphere.datasource.";private static final String DATA_SOURCE_NAME = "name";private static final String DATA_SOURCE_NAMES = "names";private static final String DATA_SOURCE_TYPE = "type";@SuppressWarnings("unchecked")public static DataSource getDataSource(final Environment environment, final String dataSourceName) throws NamingException {Binder binder = Binder.get(environment);BindResult<Map> bind = binder.bind(toDashedForm(String.join("", PREFIX, dataSourceName)), Map.class);Map<String, Object> dataSourceProps = bind.get();Preconditions.checkState(!dataSourceProps.isEmpty(), "Wrong datasource [%s] properties.", dataSourceName);return DataSourcePoolCreator.create(new DataSourceProperties(dataSourceProps.get(DATA_SOURCE_TYPE).toString(),getCamelCaseKeys(dataSourceProps)));}private static String toDashedForm(final String name) {StringBuilder result = new StringBuilder(name.length());boolean inIndex = false;for (int i = 0; i < name.length(); i++) {char ch = name.charAt(i);if (inIndex) {result.append(ch);if (ch == ']') {inIndex = false;}} else {if (ch == '[') {inIndex = true;result.append(ch);} else {ch = (ch != '_') ? ch : '-';if (Character.isUpperCase(ch) && result.length() > 0 && result.charAt(result.length() - 1) != '-') {result.append('-');}result.append(Character.toLowerCase(ch));}}}return result.toString();}public static Map<String, Object> getCamelCaseKeys(final Map<String, Object> dataSourceProps) {Map<String, Object> result = new LinkedHashMap<>(dataSourceProps.size(), 1);for (Map.Entry<String, Object> entry : dataSourceProps.entrySet()) {String key = entry.getKey();result.put(key.contains("-") ? CaseFormat.LOWER_HYPHEN.to(CaseFormat.LOWER_CAMEL, key) : key, entry.getValue());}return result;}private static List<String> getDataSourceNames(final Environment environment) {StandardEnvironment standardEnv = (StandardEnvironment) environment;standardEnv.setIgnoreUnresolvableNestedPlaceholders(true);String dataSourceNames = standardEnv.getProperty(PREFIX + DATA_SOURCE_NAME);if (Strings.isNullOrEmpty(dataSourceNames)) {dataSourceNames = standardEnv.getProperty(PREFIX + DATA_SOURCE_NAMES);}return new InlineExpressionParser(dataSourceNames).splitAndEvaluate();}public static Map<String, DataSource> createDataSourceMap(Environment environment) {Map<String, DataSource> result = new LinkedHashMap<>();for (String each : getDataSourceNames(environment)) {try {result.put(each, getDataSource(environment, each));} catch (final NamingException ex) {throw new IllegalStateException(ex);}}return result;}
}MonthShardingDatabaseUtil
/*** @author wangp*/
@Slf4j
public class MonthShardingDatabaseUtil {/*** 这里使用ThreadLocal的一个目的是因为SimpleDateFormat 不是线程安全的*/private static final ThreadLocal<SimpleDateFormat> formatThreadLocal = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyyMM"));/*** 根据逻辑表从数据库刷新实际表*/public static Set<String> queryActuallyTables(Connection connection, String logicTableName) {// 使用正则表达式过滤符合条件的表 stress_data_[0-9]{4}Pattern monthPattern = Pattern.compile("^" + logicTableName + "_[0-9]{6}$");try (Statement statement = connection.createStatement()) {String sql = "SHOW TABLES LIKE '" + logicTableName + "_%';";ResultSet resultSet = statement.executeQuery(sql);Set<String> tables = new LinkedHashSet<>();while (resultSet.next()) {String tableName = resultSet.getString(1);if (monthPattern.matcher(tableName).matches()) {tables.add(tableName);}}return tables;} catch (SQLException e) {log.error("queryActuallyTables failed", e);return null;}}/*** 创建数据库表, 并开启双重验证判断表是否创建成功** @param likeTable 逻辑表名称* @param createTable 实际表名称*/public static boolean createTableLike(Connection connection, String likeTable, String createTable) {try (Statement statement = connection.createStatement()) {// create table stress_data_202210 like stress_datastatement.execute("CREATE TABLE " + createTable + " LIKE " + likeTable + ";");// double checkResultSet resultSet = statement.executeQuery("SHOW TABLES LIKE '" + createTable + "';");if (resultSet.next()) {log.info("Auto create actually table {} success", createTable);return resultSet.getString(1) != null;} else {log.info("Auto create actually table {} failure", createTable);return false;}} catch (SQLException e) {// already existsif (e.getMessage() != null && e.getMessage().contains("already exists")) {return true;}log.error("fail to create actuallyTable '{}' from logicTable '{}'",createTable, likeTable, e);return false;}}/*** 按月分表的表后缀*/public static String shardingSuffix(Date shardingValue) {return "_" + formatThreadLocal.get().format(shardingValue);}
}
四、实现按月分表算法
我们中途休息一下，总结一下上面做了那些事情呢？
- 实现了多数据源的切换，并注册到Springboot中
- 从配置文件中读取多数据源，生成ShardingSphere数据域与原生数据库的数据源，设置到动态数据源中。
- 从配置文件中读取分表的数据库表，设置分表算法，根据自动从数据库中根据逻辑表扫描分表，添加到ShardingSphere数据源中，如果没有真实表存在，根据当前时间自动创建一个分表。
1. 分表算法
通过上面的总结，基本上已经把数据源给配置完成了，现在还缺少灵魂中的灵魂，分表算法。分表算法的添加流程这里就不在详细说了，具体流程可以参考第二篇文章 ShardingSphere笔记（二）：自定义分片算法 — 按月分表。
/*** 按月分表的 Sharding 算法** @author wangp*/
@Getter
@Slf4j
public class HisDataMonthShardingAlgorithm implements StandardShardingAlgorithm<Date> {private final ThreadLocal<SimpleDateFormat> formatThreadLocal = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyyMM"));private Properties props;/*** 自动建表的下限：时间小于该时间段的就不建表了，没有意义，这个时间可以自己在配置文件中设置，不设置默认为2018年01月*/private Date autoCreateTableLowerDate;@Overridepublic void init(Properties properties) {// 读取配置文件this.props = properties;String autoCreateTableLowerDate = properties.getProperty("auto-create-table-lower");try {this.autoCreateTableLowerDate = formatThreadLocal.get().parse(autoCreateTableLowerDate);} catch (Exception e) {log.error("parse auto-create table lower date failed: {}, use default date 2018-01", e.getMessage());try {this.autoCreateTableLowerDate = formatThreadLocal.get().parse("201801");} catch (ParseException ignored) {}}}/*** 精确匹配，只有在该条件下才需要自动建表。* <p>* 因为自动建表只有在存数据的时候才有意义，查询数据的时候表没有就没有了，存数据的时候没有表可就有问题了* <p>* 而存数据都是精确匹配，所以只需要在这里自动建表** @param availableTargetNames 当前系统中可用的真实表列表* @param shardingValue 精确匹配的条件* @return 精确匹配后的表*/@Overridepublic String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<Date> shardingValue) {Date value = shardingValue.getValue();String actuallyTableName = shardingValue.getLogicTableName() + MonthShardingDatabaseUtil.shardingSuffix(value);// 从可用的列表总判断是否存在需要的表，如果表不存在，建表，并把建好的表再次添加到 availableTargetNames 中// 这里添加到 availableTargetNames 相当于刷新了ShardingSphere的缓存了，这就是一个骚操作了，后面会贴一下源码Collection<String> tables = AutoCreateTableHelper.getInstance().autoDetectionSchemeTables(shardingValue.getLogicTableName(),Collections.singletonList(actuallyTableName), true);availableTargetNames.addAll(tables);return CollectionUtil.get(tables, 0);}/*** 范围匹配，范围匹配都是查询、更新或者删除，所以表都是存在的不自动建表*/@Overridepublic Collection<String> doSharding(Collection<String> availableTargetNames, RangeShardingValue<Date> shardingValue) {// 从查询时间到现在Date time;// 判断是否有下限if (shardingValue.getValueRange().hasLowerBound()) {time = shardingValue.getValueRange().lowerEndpoint();} else {// 无下限使用默认的下限time = autoCreateTableLowerDate;}Date now;// 判断是否有上限if (shardingValue.getValueRange().hasUpperBound()) {now = shardingValue.getValueRange().upperEndpoint();} else {// 往后延一个月 （无上限的话最多往后延一个月）now = DateUtil.offsetMonth(new Date(), 1);}now = DateUtil.endOfMonth(now);List<String> tableNames = new ArrayList<>();while (time.before(now)) {String actuallyTableName = shardingValue.getLogicTableName() + MonthShardingDatabaseUtil.shardingSuffix(time);tableNames.add(actuallyTableName);time = DateUtil.offsetMonth(time, 1);}// 这里不自动建表final Collection<String> tables = AutoCreateTableHelper.getInstance().autoDetectionSchemeTables(shardingValue.getLogicTableName(), tableNames, false);availableTargetNames.addAll(tables);return tables;}/*** SPI 方式实现的分表算法，算法标识ID为 HIS_DATA_SPI_BASED*/@Overridepublic String getType() {return "HIS_DATA_SPI_BASED";}}这里分表的逻辑是，查询到对应的真实表后，判断 ShardingSphere 中是否存在该真实表，如果表不存在，自动创建该表，并更新到ShardingSphere的缓存中。
这里就是一个骚操作的地方了，为甚了新加到 availableTargetNames 中的表 ShardingSphere 就认呢？不需要修改分表配置什么的吗？原来我也是这么想的，所以写的代码贼复杂，还要有定时任务重设 ShardingSphere配置，还不成功…, 后来追源码看到了一个神器的东西。如下：
ShardingSphere 源码中的路由逻辑如下：
ShardingStandardRoutingEngine
private Collection<DataNode> routeTables(final TableRule tableRule, final String routedDataSource,final ShardingStrategy tableShardingStrategy, final List<ShardingConditionValue> tableShardingValues) {// 从 TableRule 中获取真实表内容Collection<String> availableTargetTables = tableRule.getActualTableNames(routedDataSource);Collection<String> routedTables = tableShardingValues.isEmpty()? availableTargetTables: tableShardingStrategy.doSharding(availableTargetTables, tableShardingValues, tableRule.getTableDataNode(), properties);Collection<DataNode> result = new LinkedList<>();for (String each : routedTables) {result.add(new DataNode(routedDataSource, each));}return result;
}
而 在TableRule 中，getActualTableNames() 是从一个Map中拿的值。作为缓存。
TableRule
private final Map<String, Collection<String>> dataSourceToTablesMap = new HashMap<>();public Collection<String> getActualTableNames(final String targetDataSource) {// 从缓存中拿真实表的列表return dataSourceToTablesMap.getOrDefault(targetDataSource, Collections.emptySet());}private void addActualTable(final String datasourceName, final String tableName) {// 添加真实表(框架在初始化的收自动调用)dataSourceToTablesMap.computeIfAbsent(datasourceName, key -> new LinkedHashSet<>()).add(tableName);
}
可以看到，这里存的就是真实表列表，也会就是说我们修改的那个 availableTargetNames 就是这个缓存中的对象，人家是直接把对象给你了，那你岂不是想怎么操作就怎么操作了，诶嘿。
所以我前面会说，这个骚操作还挺依赖框架本身的实现的，万一人家哪一天不爽了，不想给你对象了，给你一个拷贝对象，就抓瞎了。
这里说明了，为什么初始的时候一定要存在一个真实表，如果不存在也一定要创建一个，这就是因为如果真实表一个也没有吗，框架本身不会调用 addActualTable 函数，就是说缓存map里面没有列表，就会返回 Collections.emptySet(), 而这个就是非常坑爹的不可修改集合 ，大家学集合基础知识的时候应该都注意过这一点。它不可修改就意味着自动建表的表没办法加进去了，就会报错，所以这里我们必须保证数据库里面一定至少要有一个真实表。
2. 添加SPI
在 resource/META-INF/services/org.apache.shardingsphere.sharding.spi.ShardingAlgorithm 文件中添加
com.akxy.platdatamanage.sharding.HisDataMonthShardingAlgorithm
五、自动建表
上面有一个自动建表的辅助类，这里就只贴源码了，不多做解释了，就是从数据库查询表，然后判断表是否存在，不存在建表，更新缓存。
就是几个关键的点：
- 使用原生JDBC数据源查找数据库中存在的真实表，建表。因为ShardingSphere数据源不支持。这也是我们上面设置了两种数据源的好处，正好就可以利用上。
- 查询到的数据库表进行缓存。提升查询速度。
/*** @author wangp*/
@Slf4j
public class AutoCreateTableHelper implements EnvironmentAware {/*** 利用Springboot的单例模式，方便使用*/private static AutoCreateTableHelper instance;/*** 冬天该数据源*/private final ShardingSphereDynamicDataSource dataSource;/*** 数据库的 数据源 : 逻辑表 : 实际表 缓存*/private final Map<String, Map<String, Set<String>>> dsLogicActuallyTableCache = new ConcurrentHashMap<>();public AutoCreateTableHelper(DataSource dataSource) {this.dataSource = ((ShardingSphereDynamicDataSource) dataSource);}public Collection<String> autoDetectionSchemeTables(String logicTableName,Collection<String> actuallyTableNames, boolean createTable) {// 当前数据源String currentDataSource = getCurrentDataSource();// 从动态数据库中获取数据源的原始JDBC数据源，因为ShardingSphere数据源做不到建表语句，必须使用原始JDBC数据源try (Connection connection = getRawJdbcConnection()) {if (connection == null) {log.error("Un found raw jdbc resource to get connection for {}", currentDataSource);return Collections.emptyList();}// 获取到缓存的数据源表Set<String> cachedTables = getCachedTables(logicTableName, currentDataSource, connection);// 不存在的表List<String> notExistsTables = new ArrayList<>();// 已存在的表List<String> existsTables = new ArrayList<>();actuallyTableNames.forEach(table -> {if (cachedTables.contains(table)) {existsTables.add(table);} else {notExistsTables.add(table);}});if (notExistsTables.isEmpty()) {return existsTables;}// 如果自动建表，创建那些不存在的表if (createTable) {String tableLike = cachedTables.stream().findAny().orElse(logicTableName);for (String notExistsTable : notExistsTables) {if (MonthShardingDatabaseUtil.createTableLike(connection, tableLike, notExistsTable)) {cachedTables.add(notExistsTable);existsTables.add(notExistsTable);} else {log.info("Auto create actually table failed");}}} else {// 重新从数据库中刷新数据Set<String> databaseTables = MonthShardingDatabaseUtil.queryActuallyTables(connection, logicTableName);if (databaseTables != null) {// 刷新缓存cachedTables.addAll(databaseTables);notExistsTables.removeIf(table -> !databaseTables.contains(table));existsTables.addAll(notExistsTables);}}return existsTables;} catch (Exception e) {log.error("detection scheme table failed", e);return Collections.emptyList();}}private String getCurrentDataSource() {String dataSourceKey = DynamicDataSourceHelper.dataSourceKey();if (StrUtil.isEmpty(dataSourceKey)) {return dataSource.getDefaultDataSource();}return dataSourceKey;}private Set<String> getCachedTables(String logicTableName, String currentDataSource, Connection connection) {if (!dsLogicActuallyTableCache.containsKey(currentDataSource)) {Set<String> dataSourceTables = MonthShardingDatabaseUtil.queryActuallyTables(connection, logicTableName);if (dataSourceTables == null) {return Collections.emptySet();}Map<String, Set<String>> tableMap = new ConcurrentHashMap<>();tableMap.put(logicTableName, dataSourceTables);dsLogicActuallyTableCache.put(currentDataSource, tableMap);}Map<String, Set<String>> tableMap = dsLogicActuallyTableCache.get(currentDataSource);if (!tableMap.containsKey(logicTableName)) {Set<String> dataSourceTables = MonthShardingDatabaseUtil.queryActuallyTables(connection, logicTableName);if (dataSourceTables == null) {return Collections.emptySet();}tableMap.put(logicTableName, dataSourceTables);}return tableMap.get(logicTableName);}/*** 单例模式： 这里不使用Springboot的bean加载是为了防止作为lib的时候Bean加载时机不同导致的NPL*/public static AutoCreateTableHelper getInstance() {return instance;}Connection getRawJdbcConnection() throws Exception {String currentDataSource = getCurrentDataSource();DataSource dataSourceById;// 如果已经是 RAWJdbc的连接了，直接查找if (DynamicDataSourceHelper.isRawJdbcDataSourceKey(currentDataSource)) {dataSourceById = dataSource.getDataSourceById(currentDataSource);if (dataSourceById != null) {return dataSourceById.getConnection();}} else {// 尝试查找Jdbc连接dataSourceById = dataSource.getDataSourceById(DynamicDataSourceHelper.rawJdbcDataSourceKey(currentDataSource));if (dataSourceById != null) {return dataSourceById.getConnection();}// 如果查找不到，再次尝试使用原始连接查找dataSourceById = dataSource.getDataSourceById(currentDataSource);if (dataSourceById != null && !(dataSourceById instanceof ShardingSphereDataSource)) {return dataSourceById.getConnection();}}return null;}@Overridepublic void setEnvironment(Environment environment) {instance = this;}
}
六、 拓展
至此、所有的逻辑就已经写完了，使用的时候只需要配置配置文件即可。配置多数据源，配置那些表需要按月分表。
另外必须要提一个的就是，本项目是有缺陷的，因为基本上是为我们公司写的一个框架性项目，能满足本公司使用，额外的扩展性需求大佬们刻印扩展。
另外如果想做成 SpringbootStarter方式使用的话，做到代码的零侵入，直接引入即可使用，可以将其作为 Springboot-starter 来配置。
配置方式为：
resource/META-INF/spring.factories
org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.akxy.platdatamanage.conf.AkxyShardingSphereAutoConfiguration
或者 Springboot 2.7 之后
resource/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
com.akxy.platdatamanage.conf.AkxyShardingSphereAutoConfiguration
两种都配置上，即可兼容不同的Springboot版本。
这样的话使用的时候只需要引入依赖，配置配置文件即可，就像 各种 Springboot-starter 使用那样。
七、源代码
刚写文章的时候看到了 CSDN支持上传代码包，就把源代码放到代码包里面了，不知道这个下载是不是需要积分啥的呢，反正上面已经把所有的代码头贴上去了。实在不行到时候另起一个仓库开源出来。完事，晚安世界。
贴一个最近瞎画的可爱宵宫，嘿嘿：
ShardingSphere笔记（三）：自定义分片算法 — 按月分表·真·自动建表相关推荐
- 第五章：Sharding-JDBC 自定义分片算法
Sharding-JDBC 自定义分片算法 自定义分片算法 Sharding提供了以下4种算法接口 PreciseShardingAlgorithm RangeShardingAlgorithm Hi ...
- mysql 按月自动建表
1.开启事件 a.查询事件是否开启 SHOW VARIABLES LIKE '%scheduler%'; b.设置开启事件 set global event_scheduler ...
- Spring Boot集成ShardingSphere分片利器 AutoTable （二）—— 自动分片算法示例 | Spring Cloud 46
一.前言 在前面我们通过以下章节对ShardingSphere的AutoTable 有了基础的了解: Spring Boot集成ShardingSphere分片利器 AutoTable (一)-- 简 ...
- shardingjdbc全局表_sharding-jdbc实现按年分库按月分表
最终数据库结果如下 image.png 例如有如下sql语句 select * from ips where flowtime = '20181202'; 我们规定flowtime是我们的分片键,通过 ...
- 倍增算法入门 超详细解答+LCA+RMQ(ST表)+例题剖析
目录 一.倍增算法 二.倍增算法的应用:求LCA(最近公共祖先)附模板题 三.倍增算法的应用:RMQ 问题(ST表)附模板题 一.倍增算法 要了解倍增之前,强烈建议大家先看一下这位大佬对倍增的解释:[ ...
- Hive的三种建表方式
Hive的三种建表方式 1.直接建表法: 2.抽取(as) 建表: 3.like建表 1.直接建表法: create table movies (uid string,iid string,score ...
- 倍增算法入门 超详细解答+LCA+RMQ(ST表)+例题剖析
目录 一.倍增算法 二.倍增算法的应用:求LCA(最近公共祖先)附模板题 三.倍增算法的应用:RMQ 问题(ST表)附模板题 一.倍增算法 要了解倍增之前,强烈建议大家先看一下这位大佬对倍增的解释:[ ...
- ShardingSphere 分片算法
前言 测试ShardingSphere版本:5.2.0 下面配置中省略部分前缀,仅供参考 自动分片算法 Auto Sharding Alogorithm 自动分片算法通常用在 auto-tables ...
- 可由一个尾指针唯一确定的链表有_极客算法训练笔记(三)，链表详细图解，别再逃避了朋友...
目录 缓存引爆链表 链表单链表双向链表循环链表双向循环链表 LinkedHashMap实现LRU缓存,源码解析(JDK1.8) 算法 爬楼梯 算法 反转链表 算法 链表环检测 缓存引爆链表 存储结构 ...
最新文章
- websocket获取httpsession报NullPointerException解决办法
- weblogic10重置控制台密码
- MPX2053DP压力传感器简单测量
- RedHat Enterprise Linux 5 安装GCC过程
- TCP的ACK原理和延迟确认机制
- 【转】理解Azure订阅，账户，活动目录AD，租户等概念
- 【README3】动态规划之“找零钱”说明最优子结构怎么解决
- vs2017 linux工程设置头文件,使用Visual Studio 2017作为Linux C++开发工具
- DrawArc绘制弧线
- keepalived mysql双主架构图_MySQL数据库架构和同步复制流程
- Java设计模式之观察者模式（发布-订阅模式）
- 电梯plc的io分配_用PLC构成液体混合控制系统IO分配及梯形图编程
- 百度街景瓦片下载器1.1.5
- springboot集成openoffice实现office转PDF在线预览
- 域名被hold了怎么办？
- ALSA编程细节分析
- PCB相关知识-PCB各层的用途
- flask学习之4：图片验证码
- flutter开发核心 Future/stream/bloc
- 基于SSM在线学习网站包含课程管理-错题-收藏-考试测试-学生教师管理员58198(idea-javaweb-javaee-j2ee-springboot)