阅读量:1
Sharding-JDBC系列
2、Sharding-JDBC分库分表之SpringBoot分片策略
3、Sharding-JDBC分库分表之SpringBoot主从配置
4、SpringBoot集成Sharding-JDBC-5.3.0分库分表
5、SpringBoot集成Sharding-JDBC-5.3.0实现按月动态建表分表
前言
随着业务量的递增，项目产生海量的数据，在某些场景中，需要将数据按月存储。本篇基于Sharding-JDBC 5.3.0，分享一下按月自动建表以及分表的实现。
准备工作
创建一个数据库，创建一张表，表名为tb_order。该表作为基准表。
引入依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.7.1</version> <relativePath/> <!-- lookup parent from repository --> </parent> <modelVersion>4.0.0</modelVersion> <artifactId>Sharding-JDBC-demo2</artifactId> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>com.baomidou</groupId> <artifactId>mybatis-plus-boot-starter</artifactId> <version>3.4.1</version> </dependency> <dependency> <groupId>org.apache.shardingsphere</groupId> <artifactId>shardingsphere-jdbc-core</artifactId> <version>5.3.0</version> </dependency> <dependency> <groupId>org.yaml</groupId> <artifactId>snakeyaml</artifactId> <version>1.33</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.28</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid</artifactId> <version>1.2.6</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.22</version> <scope>compile</scope> </dependency> <!--<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <optional>true</optional> <scope>runtime</scope> </dependency>--> </dependencies> </project>1）引入shardingsphere-jdbc-core 5.3.0 的版本；
2）项目中不要引入spring-boot-devtools，否则在调试启动时，会报错；
spring-boot-devtools 会在类路径上的文件发生更改时自动重启，方便开发调试。在项目部署时，通过 java -jar 启动项目时，会自动禁用开发工具。报错的原因下面说明。
分片规则配置
4.1 application.yml
server: port: 8080 spring: main: # 处理连接池冲突 allow-bean-definition-overriding: true datasource: # shardingsphere5.3.0引入ShardingSphereDriver数据库驱动 driver-class-name: org.apache.shardingsphere.driver.ShardingSphereDriver url: jdbc:shardingsphere:classpath:sharding.yml指定分片规则的文件为sharding.yml。
4.2 sharding.yml
dataSources: order_ds: dataSourceClassName: com.zaxxer.hikari.HikariDataSource driverClassName: com.mysql.cj.jdbc.Driver url: jdbc:mysql://localhost:3306/shardingjdbctest?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false username: root password: 123456 rules: - !SHARDING tables: tb_order: #逻辑表 actualDataNodes: order_ds.tb_order #表是自动创建 keyGenerateStrategy: # 指定主键生成策略 column: order_id keyGeneratorName: snowflake tableStrategy: standard: shardingColumn: order_time #分片键 shardingAlgorithmName: custom-time-sharding shardingAlgorithms: #分片算法 custom-time-sharding: type: CLASS_BASED #自定义类 props: strategy: standard algorithmClassName: com.jingai.sharding.jdbc.algorithm.OrderTimeShardingAlgorithm #分片算法 keyGenerators: # 主键生成器 snowflake: type: SNOWFLAKE props: sql-show: true # 是否打印sql1）配置真实表为tb_order，作为分表的表前缀；
2）配置分表策略为standard标准策略，以订单创建日期为分片键；
3）配置分表算法为自定义类OrderTimeShardingAlgorithm；
分片算法OrderTimeShardingAlgorithm
package com.jingai.sharding.jdbc.algorithm; @Slf4j public class OrderTimeShardingAlgorithm implements StandardShardingAlgorithm<Date> { private static final DateFormat TABLE_SHARD_TIME_FORMAT = new SimpleDateFormat("yyyyMM"); // 表分片符号。如：tb_order_202407 private static final String TABLE_SPLIT_SYMBOL = "_"; private Properties props; @Override public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<Date> shardingValue) { String logicTableName = shardingValue.getLogicTableName(); log.info("精准分片，逻辑表名：{}，节点表名：{}", logicTableName, availableTargetNames); Date time = shardingValue.getValue(); String result = logicTableName + TABLE_SPLIT_SYMBOL + TABLE_SHARD_TIME_FORMAT.format(time); // 在配置中，只配置了逻辑表名。如果只有一个，且是逻辑表名，说明需要获取所有表名 initAvailableTargetNames(availableTargetNames, logicTableName); return getAndCreateShardingTable(logicTableName, result, availableTargetNames); } @Override public Collection<String> doSharding(Collection<String> availableTargetNames, RangeShardingValue<Date> shardingValue) { String logicTableName = shardingValue.getLogicTableName(); log.info("精准分片，逻辑表名：{}，节点表名：{}", logicTableName, availableTargetNames); // 在配置中，只配置了逻辑表名。如果只有一个，且是逻辑表名，说明需要获取所有表名 initAvailableTargetNames(availableTargetNames, logicTableName); Range<Date> valueRange = shardingValue.getValueRange(); // 如果没有最大值或最小值，则全库扫描 if(!valueRange.hasLowerBound() || !valueRange.hasUpperBound()) { return availableTargetNames; } Date min = valueRange.lowerEndpoint(); Date max = valueRange.upperEndpoint(); Set<String> rs = new HashSet<>(); while (min.compareTo(max) <= 0) { String tableName = logicTableName + "_" + TABLE_SHARD_TIME_FORMAT.format(min); rs.add(tableName); min = DateUtils.addMonths(min, 1); } return getAndCreateShardingTable(logicTableName, rs, availableTargetNames); } private void initAvailableTargetNames(Collection<String> availableTargetNames, String logicTableName) { if(availableTargetNames.size() == 1 && availableTargetNames.contains(logicTableName)) { Set<String> allTableNameBySchema = ShardingAlgorithmUtil.getAllTableNameBySchema(logicTableName); availableTargetNames.clear(); availableTargetNames.addAll(allTableNameBySchema); } } /** * 检查可用的真实表，如果表名不存在，则创建新表 * @param logicTableName 逻辑表 * @param resultTableNames 操作需要的真实表 * @param availableTargetNames 可用的真实表 * @return 分片的真实表 */ private List<String> getAndCreateShardingTable(String logicTableName, Set<String> resultTableNames, Collection<String> availableTargetNames) { return resultTableNames.stream().map(name -> getAndCreateShardingTable(logicTableName, name, availableTargetNames)).collect(Collectors.toList()); } /** * 检查可用的真实表，如果表名不存在，则创建新表 * @param logicTableName * @param resultTableName * @param availableTargetNames * @return */ private String getAndCreateShardingTable(String logicTableName, String resultTableName, Collection<String> availableTargetNames) { if(availableTargetNames.contains(resultTableName)) { return resultTableName; } boolean rs = ShardingAlgorithmUtil.createShardingTable(logicTableName, resultTableName); if(rs) { availableTargetNames.add(resultTableName); return resultTableName; } return null; } @Override public Properties getProps() { return props; } @Override public void init(Properties properties) { this.props = properties; } } 1）实现StandardShardingAlgorithm接口，重写doSharding()方法；
2）根据传入的时间分片值，解析出年月，和逻辑表组合，为实际操作的真实表；
3）如果当前的真实表不存在，则调用工具类ShardingAlgorithmUtil创建一个真实表；
工具类ShardingAlgorithmUtil
package com.jingai.sharding.jdbc.util; @Slf4j public class ShardingAlgorithmUtil { // 表分片符号。如：tb_order_202407 private static final String TABLE_SPLIT_SYMBOL = "_"; // 配置的数据库源 private volatile static Map<String, Map<String, Object>> dataSources = null; public static void init(String url) { Assert.hasText(url, "分片策略不能为空"); log.info("数据源获取..."); byte[] bytes = new ShardingSphereDriverURL(url).toConfigurationBytes(); try { YamlRootConfiguration yamlRootConfiguration = YamlEngine.unmarshal(bytes, YamlRootConfiguration.class); dataSources = yamlRootConfiguration.getDataSources(); } catch(Exception e) { e.printStackTrace(); log.error("分片策略配置解析失败"); throw new IllegalArgumentException("分片策略解析失败"); } } /** * 获取所有真实表名 */ public static Set<String> getAllTableNameBySchema(String logicTableName) { Assert.notNull(dataSources, "分片策略配置未初始化"); Set<String> rs = new HashSet<>(); // 获取配置的数据源 String startTable = logicTableName + TABLE_SPLIT_SYMBOL; for (Map<String, Object> dataSource : dataSources.values()) { try (Connection conn = DriverManager.getConnection(dataSource.get("url").toString(), dataSource.get("username").toString(), dataSource.get("password").toString())){ Statement statement = conn.createStatement(); ResultSet resultSet = statement.executeQuery("show tables like '" + startTable + "%'"); while (resultSet.next()) { String tableName = resultSet.getString(1); if(StringUtils.hasText(tableName) && tableName.replaceFirst(startTable, "").matches("\\d{6}")) { rs.add(tableName); } } } catch(Exception e) { e.printStackTrace(); throw new IllegalArgumentException("数据库连接失败"); } } return rs; } /** * 创建分表 * @param logicTableName * @param resultTableName * @return */ public static boolean createShardingTable(String logicTableName, String resultTableName) { synchronized (logicTableName.intern()) { for (Map<String, Object> dataSource : dataSources.values()) { try (Connection conn = DriverManager.getConnection(dataSource.get("url").toString(), dataSource.get("username").toString(), dataSource.get("password").toString())){ Statement statement = conn.createStatement(); log.info("创建{}表", resultTableName); statement.execute("create table if not exists `" + resultTableName + "` like `" + logicTableName + "`;"); } catch(Exception e) { e.printStackTrace(); throw new IllegalArgumentException("数据库连接失败"); } } return true; } } } 1）init(String url) 初始化方法，通过传入的url（application.yml中配置的spring.datasource.url），解析分片配置文件，得到配置的datasources信息；
2）getAllTableNameBySchema(String logicTableName)，通过传入的逻辑表（配置中的tb_order），结合配置的datasources信息，创建连接，从数据库中获取表名以tb_order为前缀的表。即数据库中的真实表；
真实表只需从主库中获取即可，此处可以完善。
3）createShardingTable()，结合配置的datasources信息，创建连接，创建真实表；
初始化类
package com.jingai.sharding.jdbc.listener; @Component @Slf4j public class ShardingInitRunner implements InitializingBean { @Value("${spring.datasource.url}") private String url; @Override public void afterPropertiesSet() throws Exception { log.info("sharding初始化..."); ShardingAlgorithmUtil.init(url); } } 该类获取spring.datasource.url的配置值，在初始化方法中，调用ShardingAlgorithmUtil.init(url)，初始化ShardingAlgorithmUtil中的datasource值。
1）如果引入了spring-boot-devtools依赖，开启开发工具。项目启动的时候，ShardingAlgorithmUtil类的类加载器为devtools包下的RestartClassLoader，并执行了初始化，获取了datasources；
2）在分片算法OrderTimeShardingAlgorithm的类加载器为AppClassLoader，OrderTimeShardingAlgorithm中调用ShardingAlgorithmUtil时，会用AppClassLoader重新加载一次ShardingAlgorithmUtil，此时的datasources为null；
3）此时执行ShardingAlgorithmUtil操作数据库时，会报空指针；
实体类
package com.jingai.sharing.jdbc.entity; @Data @ToString @TableName("tb_order") public class OrderEntity { private long orderId; private long memberId; private float totalPrice; private String status; private Date orderTime; }在实体类中，@TableName指定配置中的逻辑表。
Mapper类
package com.jingai.sharing.jdbc.dao; public interface OrderMapper extends BaseMapper<OrderEntity> { @Insert("insert into tb_order(member_id, total_price, status, order_time) values " + "(#{memberId}, #{totalPrice}, #{status}, #{orderTime})") @Options(useGeneratedKeys = true, keyProperty = "orderId") int insert2(OrderEntity order); } 在4.2的配置中，通过key-generator设置了逻辑表的主键生成策略为雪花算法。当进行数据插入时，需要编写新的插入接口，不能直接使用Mybatis-plus中的insert()接口。因为在默认的insert()接口中，实体对象的orderId为0，不会走配置的雪花算法。
Service类
package com.jingai.sharing.jdbc.service; @Service public class OrderService extends ServiceImpl<OrderMapper, OrderEntity> { @Resource private OrderMapper orderMapper; public long insert2(OrderEntity order) { int rs = orderMapper.insert2(order); return rs > 0 ? order.getOrderId() : 0; } } 为了便于测试，此处省略了Service的接口类。
Controller类
@RestController public class OrderController { @Resource private OrderService orderService; @RequestMapping("order") public String order(OrderEntity order) { order.setOrderTime(new Date()); long insert = orderService.insert2(order); return insert > 0 ? "success" : "fail"; } @RequestMapping("list") public List<OrderEntity> list() { return orderService.list(); } }小结
以上为本篇分享的全部内容。以下做一个小结：
1）创建一个基准表tb_order；
2）配置分片规则：标准策略、以订单时间为分片键、自定义分片算法；
3）在分片算法中，根据分片键的值日期值，找到对应月份的表。如果真实表不存在，则创建；
关于本篇内容你有什么自己的想法或独到见解，欢迎在评论区一起交流探讨下吧。
