中小企业网站建设与推广分析,wordpress thegem,免费企业一键建站网站,手机网站主页推荐ShardingSphere笔记#xff08;二#xff09;#xff1a;自定义分片算法 — 按月分表真自动建表 文章目录 ShardingSphere笔记#xff08;二#xff09;#xff1a;自定义分片算法 — 按月分表真自动建表一、 前言二、 Springboot 的动态数据库三、 实现我们自己的动态数…ShardingSphere笔记二自定义分片算法 — 按月分表·真·自动建表 文章目录 ShardingSphere笔记二自定义分片算法 — 按月分表·真·自动建表一、 前言二、 Springboot 的动态数据库三、 实现我们自己的动态数据库1. POM2. 实现 AbstractRoutingDatasource3. 创建数据源 四、实现按月分表算法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 {/*** pAttempts 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 {/*** 配置多个数据源 * keystring唯一标识数据源* value: datasource 数据源 */Nullableprivate MapObject, DataSource resolvedDataSources;public Connection getConnection() throws SQLException {return this.determineTargetDataSource().getConnection();}/*** 确定当前的数据源*/protected DataSource determineTargetDataSource() {// 获取数据库IdkeyObject 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做配置文件提示的 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-configuration-processor/artifactIdoptionaltrue/optional/dependency!-- Lombok 开发神器懂得都懂 --dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependency!-- ShardingSphere 核心依赖 --dependencygroupIdorg.apache.shardingsphere/groupIdartifactIdshardingsphere-jdbc-core/artifactIdversion${shardingsphere-jdbc.version}/version/dependencydependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-jdbc/artifactId/dependency!-- 工具类库有点大看需求引入只是用习惯了 --dependencygroupIdcn.hutool/groupIdartifactIdhutool-all/artifactIdversion5.8.11/version/dependency!-- 阿里的 Transmittable ThreadLocal 解决父子线程值传递问题 --dependencygroupIdcom.alibaba/groupIdartifactIdtransmittable-thread-local/artifactIdversion2.14.2/version/dependency2. 实现 AbstractRoutingDatasource
ShardingSphereDynamicDataSource
public class ShardingSphereDynamicDataSource extends AbstractRoutingDataSource {/*** 多数据库 Map* 这里重新创建了一个Map对象目的有两个* 1. ards 的 resolvedDataSources 是 private 的子类无法访问。* 2. 设置为static方式的可以通过静态方法访问更方便。*/static final MapObject, Object DATA_SOURCE_MAP new ConcurrentHashMap(10);private String defaultDataSource;public ShardingSphereDynamicDataSource(LinkedHashMapString, 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 ThreadLocalString 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?useSSLfalseuseUnicodetruecharacterEncodingutf-8zeroDateTimeBehaviorconvertToNullserverTimezoneAsia/ShanghaiallowMultiQueriestrueusername: rootpassword: 123456# 数据2ds2:type: com.zaxxer.hikari.HikariDataSourcedriver-class-name: com.mysql.jdbc.Driverjdbc-url: jdbc:mysql://127.0.0.1:3306/test2?useSSLfalseuseUnicodetruecharacterEncodingutf-8zeroDateTimeBehaviorconvertToNullserverTimezoneAsia/ShanghaiallowMultiQueriestrueusername: 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 ListString 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());// 从配置文件中读取数据源MapString, DataSource dataSourceMap DatasourceHelper.createDataSourceMap(environment);LinkedHashMapString, 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 {SetString 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);BindResultMap bind binder.bind(toDashedForm(String.join(, PREFIX, dataSourceName)), Map.class);MapString, 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 MapString, Object getCamelCaseKeys(final MapString, Object dataSourceProps) {MapString, Object result new LinkedHashMap(dataSourceProps.size(), 1);for (Map.EntryString, 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 ListString 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 MapString, DataSource createDataSourceMap(Environment environment) {MapString, 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 ThreadLocalSimpleDateFormat formatThreadLocal ThreadLocal.withInitial(() - new SimpleDateFormat(yyyyMM));/*** 根据逻辑表从数据库刷新实际表*/public static SetString 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);SetString 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 StandardShardingAlgorithmDate {private final ThreadLocalSimpleDateFormat 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(CollectionString availableTargetNames, PreciseShardingValueDate shardingValue) {Date value shardingValue.getValue();String actuallyTableName shardingValue.getLogicTableName() MonthShardingDatabaseUtil.shardingSuffix(value);// 从可用的列表总判断是否存在需要的表如果表不存在建表并把建好的表再次添加到 availableTargetNames 中// 这里添加到 availableTargetNames 相当于刷新了ShardingSphere的缓存了这就是一个骚操作了后面会贴一下源码CollectionString tables AutoCreateTableHelper.getInstance().autoDetectionSchemeTables(shardingValue.getLogicTableName(),Collections.singletonList(actuallyTableName), true);availableTargetNames.addAll(tables);return CollectionUtil.get(tables, 0);}/*** 范围匹配范围匹配都是查询、更新或者删除所以表都是存在的不自动建表*/Overridepublic CollectionString doSharding(CollectionString availableTargetNames, RangeShardingValueDate 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);ListString tableNames new ArrayList();while (time.before(now)) {String actuallyTableName shardingValue.getLogicTableName() MonthShardingDatabaseUtil.shardingSuffix(time);tableNames.add(actuallyTableName);time DateUtil.offsetMonth(time, 1);}// 这里不自动建表final CollectionString 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 CollectionDataNode routeTables(final TableRule tableRule, final String routedDataSource,final ShardingStrategy tableShardingStrategy, final ListShardingConditionValue tableShardingValues) {// 从 TableRule 中获取真实表内容CollectionString availableTargetTables tableRule.getActualTableNames(routedDataSource);CollectionString routedTables tableShardingValues.isEmpty()? availableTargetTables: tableShardingStrategy.doSharding(availableTargetTables, tableShardingValues, tableRule.getTableDataNode(), properties);CollectionDataNode result new LinkedList();for (String each : routedTables) {result.add(new DataNode(routedDataSource, each));}return result;
}而 在TableRule 中getActualTableNames() 是从一个Map中拿的值。作为缓存。
TableRule private final MapString, CollectionString dataSourceToTablesMap new HashMap();public CollectionString 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 MapString, MapString, SetString dsLogicActuallyTableCache new ConcurrentHashMap();public AutoCreateTableHelper(DataSource dataSource) {this.dataSource ((ShardingSphereDynamicDataSource) dataSource);}public CollectionString autoDetectionSchemeTables(String logicTableName,CollectionString 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();}// 获取到缓存的数据源表SetString cachedTables getCachedTables(logicTableName, currentDataSource, connection);// 不存在的表ListString notExistsTables new ArrayList();// 已存在的表ListString 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 {// 重新从数据库中刷新数据SetString 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 SetString getCachedTables(String logicTableName, String currentDataSource, Connection connection) {if (!dsLogicActuallyTableCache.containsKey(currentDataSource)) {SetString dataSourceTables MonthShardingDatabaseUtil.queryActuallyTables(connection, logicTableName);if (dataSourceTables null) {return Collections.emptySet();}MapString, SetString tableMap new ConcurrentHashMap();tableMap.put(logicTableName, dataSourceTables);dsLogicActuallyTableCache.put(currentDataSource, tableMap);}MapString, SetString tableMap dsLogicActuallyTableCache.get(currentDataSource);if (!tableMap.containsKey(logicTableName)) {SetString 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.EnableAutoConfigurationcom.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支持上传代码包就把源代码放到代码包里面了不知道这个下载是不是需要积分啥的呢反正上面已经把所有的代码头贴上去了。实在不行到时候另起一个仓库开源出来。完事晚安世界。
贴一个最近瞎画的可爱宵宫嘿嘿
文章转载自： http://www.morning.jjzbx.cn.gov.cn.jjzbx.cn http://www.morning.mrtdq.cn.gov.cn.mrtdq.cn http://www.morning.jgcyn.cn.gov.cn.jgcyn.cn http://www.morning.ljdd.cn.gov.cn.ljdd.cn http://www.morning.rsjf.cn.gov.cn.rsjf.cn http://www.morning.tdhxp.cn.gov.cn.tdhxp.cn http://www.morning.tymnr.cn.gov.cn.tymnr.cn http://www.morning.qfqld.cn.gov.cn.qfqld.cn http://www.morning.bdtpd.cn.gov.cn.bdtpd.cn http://www.morning.qjbxt.cn.gov.cn.qjbxt.cn http://www.morning.mslsn.cn.gov.cn.mslsn.cn http://www.morning.txjrc.cn.gov.cn.txjrc.cn http://www.morning.lxqkt.cn.gov.cn.lxqkt.cn http://www.morning.pxmyw.cn.gov.cn.pxmyw.cn http://www.morning.fssmx.com.gov.cn.fssmx.com http://www.morning.cflxx.cn.gov.cn.cflxx.cn http://www.morning.btqrz.cn.gov.cn.btqrz.cn http://www.morning.pcqxr.cn.gov.cn.pcqxr.cn http://www.morning.klrpm.cn.gov.cn.klrpm.cn http://www.morning.hrzhg.cn.gov.cn.hrzhg.cn http://www.morning.txtzr.cn.gov.cn.txtzr.cn http://www.morning.atoinfo.com.gov.cn.atoinfo.com http://www.morning.rshs.cn.gov.cn.rshs.cn http://www.morning.wnrcj.cn.gov.cn.wnrcj.cn http://www.morning.qfths.cn.gov.cn.qfths.cn http://www.morning.ghpld.cn.gov.cn.ghpld.cn http://www.morning.qrwdg.cn.gov.cn.qrwdg.cn http://www.morning.syxmx.cn.gov.cn.syxmx.cn http://www.morning.pwppk.cn.gov.cn.pwppk.cn http://www.morning.xtdtt.cn.gov.cn.xtdtt.cn http://www.morning.ljzgf.cn.gov.cn.ljzgf.cn http://www.morning.kmcby.cn.gov.cn.kmcby.cn http://www.morning.pghfy.cn.gov.cn.pghfy.cn http://www.morning.fslxc.cn.gov.cn.fslxc.cn http://www.morning.hjrjy.cn.gov.cn.hjrjy.cn http://www.morning.tkryt.cn.gov.cn.tkryt.cn http://www.morning.trrrm.cn.gov.cn.trrrm.cn http://www.morning.xbnkm.cn.gov.cn.xbnkm.cn http://www.morning.mzbyl.cn.gov.cn.mzbyl.cn http://www.morning.ghrhb.cn.gov.cn.ghrhb.cn http://www.morning.ldsgm.cn.gov.cn.ldsgm.cn http://www.morning.lwxsy.cn.gov.cn.lwxsy.cn http://www.morning.bzlfw.cn.gov.cn.bzlfw.cn http://www.morning.bzgpj.cn.gov.cn.bzgpj.cn http://www.morning.fbtgp.cn.gov.cn.fbtgp.cn http://www.morning.jsdntd.com.gov.cn.jsdntd.com http://www.morning.tgyzk.cn.gov.cn.tgyzk.cn http://www.morning.myfwb.cn.gov.cn.myfwb.cn http://www.morning.shnqh.cn.gov.cn.shnqh.cn http://www.morning.rlsd.cn.gov.cn.rlsd.cn http://www.morning.stcds.cn.gov.cn.stcds.cn http://www.morning.ktblf.cn.gov.cn.ktblf.cn http://www.morning.ktdqu.cn.gov.cn.ktdqu.cn http://www.morning.ybmp.cn.gov.cn.ybmp.cn http://www.morning.hcgbm.cn.gov.cn.hcgbm.cn http://www.morning.wrtw.cn.gov.cn.wrtw.cn http://www.morning.gwzfj.cn.gov.cn.gwzfj.cn http://www.morning.hmtft.cn.gov.cn.hmtft.cn http://www.morning.lqffg.cn.gov.cn.lqffg.cn http://www.morning.muniubangcaishui.cn.gov.cn.muniubangcaishui.cn http://www.morning.thntp.cn.gov.cn.thntp.cn http://www.morning.nqbs.cn.gov.cn.nqbs.cn http://www.morning.wffxr.cn.gov.cn.wffxr.cn http://www.morning.rqnhf.cn.gov.cn.rqnhf.cn http://www.morning.hgscb.cn.gov.cn.hgscb.cn http://www.morning.yrfxb.cn.gov.cn.yrfxb.cn http://www.morning.qlznd.cn.gov.cn.qlznd.cn http://www.morning.kphyl.cn.gov.cn.kphyl.cn http://www.morning.kfwqd.cn.gov.cn.kfwqd.cn http://www.morning.lpgw.cn.gov.cn.lpgw.cn http://www.morning.lwcqh.cn.gov.cn.lwcqh.cn http://www.morning.zbhfs.cn.gov.cn.zbhfs.cn http://www.morning.xlbyx.cn.gov.cn.xlbyx.cn http://www.morning.drcnf.cn.gov.cn.drcnf.cn http://www.morning.zmpqt.cn.gov.cn.zmpqt.cn http://www.morning.ghfmd.cn.gov.cn.ghfmd.cn http://www.morning.kdhrf.cn.gov.cn.kdhrf.cn http://www.morning.ypfw.cn.gov.cn.ypfw.cn http://www.morning.tnmmp.cn.gov.cn.tnmmp.cn http://www.morning.sgbss.cn.gov.cn.sgbss.cn
