Sharding-JDBC动态水平分表实现

背景：

在项目中遇到了按照日期动态水平分表的需求，系统属于监控系统，每10分钟保存一次监控数据，并且每次要采集200个节点上的数据，即每次采集数据(间隔10分钟)，向数据表添加200条记录，这样一个月数据表就有将近100万条记录。

为了控制单表数据量，并且为了方便后期数据统计，所以，每个月创建一张新表，之后的采集数据都写到新表中，例如报警信息表：alarm_histrtory表，按照日期水平分表alarm_histrtory201912，alarm_histrtory202001，alarm_histrtory202002，分别代表2019年12月、2020年1月、2020年2月的数据表。

每个月都需要创建一张新的数据表，但是Sharding-JDBC水平分表不能动态变化，所以，为了实现Sharding-JDBC的水平分表配置随着时间，动态修改，而无需程序重启。

示例代码：https://github.com/xujingle1995/sharding-jdbc

解决方案：

方案一：

通过配置中心，修改配置文件，然后sharding-jdbc自动获取新的分库分表配置，从而实现动态修改。这个方案还是需要人的介入，如果需要了解这种方案，只需要springboot中引入nacos配置中心即可。如果nacos不会配置可以参考：

1.官方文档：https://github.com/alibaba/spring-cloud-alibaba/wiki/Nacos-config

2.相关博客：https://blog.csdn.net/qq_26932225/article/details/86536837、https://blog.csdn.net/qq_26932225/article/details/86548829

方案二：

只需要以下三步：

1.自定义分片算法类

2.添加spring定时任务，动态修改Sharding-JDBC的配置。

3.配置application.properties配置文件

开始实践

1.准备工作：

创建Springboot工程,pom.xml引入mysql、mybatis、sharding-jdbc依赖

创建数据库：见本文最后的SQL语句

org.springframework.boot

spring-boot-starter-web

org.mybatis.spring.boot

mybatis-spring-boot-starter

2.1.2

org.springframework.boot

spring-boot-devtools

runtime

true

mysql

mysql-connector-java

runtime

org.projectlombok

lombok

true

org.springframework.boot

spring-boot-starter-test

test

org.apache.shardingsphere

sharding-jdbc-spring-boot-starter

4.0.0-RC1

2.自定义分片算法类(重要)

自定义分片算法类，用于当SQL语句中包含了分片键，sharding-jdbc会调用该类的doSharding方法，得到要查询的实际数据表名我这里自定义乐standard的精确分片和范围分片：

package com.xjl.sharding.config;

import lombok.extern.slf4j.Slf4j;

import org.apache.shardingsphere.api.sharding.standard.PreciseShardingAlgorithm;

import org.apache.shardingsphere.api.sharding.standard.PreciseShardingValue;

import java.text.SimpleDateFormat;

import java.util.Collection;

import java.util.Date;

/**

* @Title: dongfangdianqi

* @description: alarmHistory 精确分片 = in

* @create: 2020-02-25 14:12

* @update: 2020-02-25 14:12

* @updateRemark: 修改内容

* @Version: 1.0

*/

@Slf4j

public class PreciseSharingTableAlgorithmOfAlarmhis implements PreciseShardingAlgorithm {

private SimpleDateFormat dateformat = new SimpleDateFormat("yyyyMM");

@Override

public String doSharding(Collection availableTargetNames, PreciseShardingValue shardingValue) {

StringBuffer tableName = new StringBuffer();

log.info("执行操作的表名{}",shardingValue.getLogicTableName() + dateformat.format(shardingValue.getValue()));

tableName.append(shardingValue.getLogicTableName()).append(dateformat.format(shardingValue.getValue()));

return tableName.toString();

}

}

package com.xjl.sharding.config;

import com.google.common.collect.Lists;

import com.google.common.collect.Range;

import lombok.extern.slf4j.Slf4j;

import org.apache.shardingsphere.api.sharding.standard.RangeShardingAlgorithm;

import org.apache.shardingsphere.api.sharding.standard.RangeShardingValue;

import java.text.SimpleDateFormat;

import java.util.*;

/**

* @Title: dongfangdianqi

* @description: 根据发生时间的范围查询分片算法 between and

* @create: 2020-02-25 16:55

* @update: 2020-02-25 16:55

* @updateRemark: 修改内容

* @Version: 1.0

*/

@Slf4j

public class RangeShardingAlgorithmOfAlarmhis implements RangeShardingAlgorithm {

private static SimpleDateFormat dateformat = new SimpleDateFormat("yyyyMM");

@Override

public Collection doSharding(Collection availableTargetNames, RangeShardingValue shardingValue) {

Collection result = new LinkedHashSet<>();

Range shardingKey = shardingValue.getValueRange();

// 获取起始，终止时间范围

Date startTime = shardingKey.lowerEndpoint();

Date endTime = shardingKey.upperEndpoint();

Date now = new Date();

if (startTime.after(now)){

startTime = now;

}

if (endTime.after(now)){

endTime = now;

}

Collection tables = getRoutTable(shardingValue.getLogicTableName(), startTime, endTime);

if (tables != null && tables.size() >0) {

result.addAll(tables);

}

return result;

}

private Collection getRoutTable(String logicTableName, Date startTime, Date endTime) {

Set rouTables = new HashSet<>();

if (startTime != null && endTime != null) {

List rangeNameList = getRangeNameList(startTime, endTime);

for (String YearMonth : rangeNameList) {

rouTables.add(logicTableName + YearMonth);

}

}

return rouTables;

}

private static List getRangeNameList(Date startTime, Date endTime) {

List result = Lists.newArrayList();

// 定义日期实例

Calendar dd = Calendar.getInstance();

dd.setTime(startTime);

while(dd.getTime().before(endTime)) {

result.add(dateformat.format(dd.getTime()));

// 进行当前日期按月份 + 1

dd.add(Calendar.MONTH, 1);

}

return result;

}

}

3.添加定时任务类

package com.xjl.sharding.config;

import lombok.Data;

import org.springframework.boot.context.properties.ConfigurationProperties;

/**

* @Description:

* @author: 许京乐

* @date: 2020/3/1 21:50

*/

@ConfigurationProperties(prefix = "dynamic.table")

@Data

public class DynamicTablesProperties {

String[] names;

}

package com.xjl.sharding.config;

import lombok.extern.slf4j.Slf4j;

import org.apache.shardingsphere.core.exception.ShardingConfigurationException;

import org.apache.shardingsphere.core.rule.DataNode;

import org.apache.shardingsphere.core.rule.TableRule;

import org.apache.shardingsphere.shardingjdbc.jdbc.core.datasource.ShardingDataSource;

import org.springframework.beans.factory.InitializingBean;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.context.properties.EnableConfigurationProperties;

import org.springframework.scheduling.annotation.EnableScheduling;

import org.springframework.scheduling.annotation.Scheduled;

import org.springframework.stereotype.Component;

import javax.sql.DataSource;

import java.lang.reflect.Field;

import java.lang.reflect.Modifier;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.ArrayList;

import java.util.List;

import java.util.Random;

/**

* @Description:水平分表，动态分表刷新定时任务

* @author: 许京乐

* @date: 2020/2/29 23:47

*/

@Component

@EnableScheduling

@EnableConfigurationProperties(DynamicTablesProperties.class)

@Slf4j

public class ShardingTableRuleActualTablesRefreshSchedule implements InitializingBean {

@Autowired

private DynamicTablesProperties dynamicTables;

@Autowired

private DataSource dataSource;

public ShardingTableRuleActualTablesRefreshSchedule() {

}

@Scheduled(cron = "0 0 0 * * *")

public void actualTablesRefresh() throws NoSuchFieldException, IllegalAccessException {

System.out.println("---------------------------------");

ShardingDataSource dataSource = (ShardingDataSource) this.dataSource;

if (dynamicTables.getNames() == null || dynamicTables.getNames().length == 0) {

log.warn("dynamic.table.names为空");

return;

}

for (int i = 0; i < dynamicTables.getNames().length; i++) {

TableRule tableRule = null;

try {

tableRule = dataSource.getShardingContext().getShardingRule().getTableRule(dynamicTables.getNames()[i]);

System.out.println(tableRule);

} catch (ShardingConfigurationException e) {

log.error("逻辑表：{},不存在配置！", dynamicTables.getNames()[i]);

}

List dataNodes = tableRule.getActualDataNodes();

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

// ！！！！！！!！默认水平分表开始时间是2019-12月，每个月新建一张新表！！！！！

LocalDateTime localDateTime = LocalDateTime.of(2019, 12, 1, 0, 0, new Random().nextInt(59));

LocalDateTime now = LocalDateTime.now();

String dataSourceName = dataNodes.get(0).getDataSourceName();

String logicTableName = tableRule.getLogicTable();

StringBuilder stringBuilder = new StringBuilder(10).append(dataSourceName).append(".").append(logicTableName);

final int length = stringBuilder.length();

List newDataNodes = new ArrayList<>();

while (true) {

stringBuilder.setLength(length);

stringBuilder.append(localDateTime.format(DateTimeFormatter.ofPattern("yyyyMM")));

DataNode dataNode = new DataNode(stringBuilder.toString());

newDataNodes.add(dataNode);

localDateTime = localDateTime.plusMonths(1L);

if (localDateTime.isAfter(now)) {

break;

}

}

actualDataNodesField.set(tableRule, newDataNodes);

}

}

@Override

public void afterPropertiesSet() throws Exception {

actualTablesRefresh();

}

}

4.application.properties配置文件

# sharding-jdbc 相关配置

# 配置水平分表随着日期每月递增的逻辑表名，配置后不走分片建，全局查询时能够自动获取最新的逻辑表分片，多个通过逗号分隔

dynamic.table.names=alarmhis

# 数据源配置

spring.shardingsphere.datasource.names = ds0

spring.shardingsphere.datasource.ds0.type = com.zaxxer.hikari.HikariDataSource

spring.shardingsphere.datasource.ds0.driver‐class‐name = com.mysql.cj.jdbc.Driver

spring.shardingsphere.datasource.ds0.jdbc-url = jdbc:mysql://IP地址:端口号/dfdq?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&useSSL=false

spring.shardingsphere.datasource.ds0.username = 你的数据库账户

spring.shardingsphere.datasource.ds0.password = 你的数据库密码

## 分表策略 其中alarmhis为逻辑表 分表主要取决与almhappentime字段

spring.shardingsphere.sharding.tables.alarmhis.actual-data-nodes=ds0.alarmhis

spring.shardingsphere.sharding.tables.alarmhis.table-strategy.standard.sharding-column=AlmClearTime

# 自定义分表算法

spring.shardingsphere.sharding.tables.alarmhis.table-strategy.standard.precise-algorithm-class-name=com.dfdq.common.sharding.jdbc.PreciseSharingTableAlgorithmOfAlarmhis

spring.shardingsphere.sharding.tables.alarmhis.table-strategy.standard.range-algorithm-class-name=com.dfdq.common.sharding.jdbc.RangeShardingAlgorithmOfAlarmhis

# 打印解析后的SQL语句

spring.shardingsphere.props.sql.show = true

# sharding jdbc 需要重新注入数据源，覆盖原本注入的数据源

spring.main.allow-bean-definition-overriding=true

5.创建实体类以及Dao层

package com.xjl.sharding.entity;

import com.fasterxml.jackson.annotation.JsonFormat;

import lombok.Data;

import java.time.Instant;

/**

* @Description:

* @date: 2020/2/29 16:11

*/

@Data

public class AlarmHistoryDO {

private int turbineID;

private int almPointID;

@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")

private Instant almHappenTime;

@JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")

private Instant almClearTime;

}

package com.xjl.sharding.dao;

import com.xjl.sharding.entity.AlarmHistoryDO;

import org.apache.ibatis.annotations.Mapper;

import org.apache.ibatis.annotations.Param;

import java.time.Instant;

import java.util.List;

/**

* @Description:历史报警信息表数据类

* @author: 许京乐

* @date: 2020/2/29 16:06

*/

@Mapper

public interface TestDao {

List getAlarmHistoryById(@Param("id") String id);

List getAlarmHistoryByTime(@Param("startTime") Instant startTime, @Param("endTime") Instant endTime);

}

6.编写单元测试

测试中SQL语句没有走分片键，实际查询语句是全表查询，并且定时任务会动态修改实际水平分表

package com.xjl.sharding;

import com.xjl.sharding.dao.TestDao;

import org.junit.Test;

import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)

@SpringBootTest

public class ShardingApplicationTests {

@Autowired

TestDao testDao;

@Test

public void contextLoads() {

testDao.getAlarmHistoryById("1");

}

}

7. 打印日志

. ____ _ __ _ _

/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \

( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \

\\/ ___)| |_)| | | | | || (_| | ) ) ) )

' |____| .__|_| |_|_| |_\__, | / / / /

=========|_|==============|___/=/_/_/_/

:: Spring Boot :: (v2.1.13.RELEASE)

2020-03-12 12:30:16.129 INFO 88264 --- [ main] c.xjl.sharding.ShardingApplicationTests : Starting ShardingApplicationTests on LAPTOP-47DUEIIO with PID 88264 (started by xujingle in D:\ProjectCode\NewDuty\sharding)

2020-03-12 12:30:16.131 INFO 88264 --- [ main] c.xjl.sharding.ShardingApplicationTests : No active profile set, falling back to default profiles: default

2020-03-12 12:30:18.437 INFO 88264 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...

2020-03-12 12:30:18.965 INFO 88264 --- [ main] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.

---------------------------------

TableRule(logicTable=alarmhis, actualDataNodes=[DataNode(dataSourceName=ds0, tableName=alarmhis201912), DataNode(dataSourceName=ds0, tableName=alarmhis202001)], databaseShardingStrategy=null, tableShardingStrategy=org.apache.shardingsphere.core.strategy.route.standard.StandardShardingStrategy@6d303498, generateKeyColumn=null, shardingKeyGenerator=null, logicIndex=null)

2020-03-12 12:30:26.022 INFO 88264 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Initializing ExecutorService 'applicationTaskExecutor'

2020-03-12 12:30:26.762 INFO 88264 --- [ main] o.s.s.c.ThreadPoolTaskScheduler : Initializing ExecutorService 'taskScheduler'

2020-03-12 12:30:26.893 INFO 88264 --- [ main] c.xjl.sharding.ShardingApplicationTests : Started ShardingApplicationTests in 11.374 seconds (JVM running for 12.764)

2020-03-12 12:30:27.990 INFO 88264 --- [ main] ShardingSphere-SQL : Rule Type: sharding

2020-03-12 12:30:27.990 INFO 88264 --- [ main] ShardingSphere-SQL : Logic SQL: SELECT * FROM alarmhis WHERE TurbineID = ?

2020-03-12 12:30:27.991 INFO 88264 --- [ main] ShardingSphere-SQL : SQLStatement: SelectStatement(super=DQLStatement(super=AbstractSQLStatement(type=DQL, tables=Tables(tables=[Table(name=alarmhis, alias=Optional.absent())]), routeConditions=Conditions(orCondition=OrCondition(andConditions=[])), encryptConditions=Conditions(orCondition=OrCondition(andConditions=[])), sqlTokens=[TableToken(tableName=alarmhis, quoteCharacter=NONE, schemaNameLength=0)], parametersIndex=1, logicSQL=SELECT * FROM alarmhis WHERE TurbineID = ?)), containStar=true, firstSelectItemStartIndex=7, selectListStopIndex=7, groupByLastIndex=0, items=[StarSelectItem(owner=Optional.absent())], groupByItems=[], orderByItems=[], limit=null, subqueryStatement=null, subqueryStatements=[], subqueryConditions=[])

2020-03-12 12:30:27.991 INFO 88264 --- [ main] ShardingSphere-SQL : Actual SQL: ds0 ::: SELECT * FROM alarmhis201912 WHERE TurbineID = ? ::: [1]

2020-03-12 12:30:27.991 INFO 88264 --- [ main] ShardingSphere-SQL : Actual SQL: ds0 ::: SELECT * FROM alarmhis202001 WHERE TurbineID = ? ::: [1]

2020-03-12 12:30:27.991 INFO 88264 --- [ main] ShardingSphere-SQL : Actual SQL: ds0 ::: SELECT * FROM alarmhis202002 WHERE TurbineID = ? ::: [1]

2020-03-12 12:30:27.991 INFO 88264 --- [ main] ShardingSphere-SQL : Actual SQL: ds0 ::: SELECT * FROM alarmhis202003 WHERE TurbineID = ? ::: [1]

8.注意的点：

com.xjl.sharding.config.ShardingTableRuleActualTablesRefreshSchedule定时任务类中，设置了默认起始分表时间是从2019-12月，每个月分表一次。

application.properties配置文件中，spring.shardingsphere.sharding.tables.alarmhis.actual-data-nodes=只需要等于逻辑表名并且dynamic.table.names也需要设置水平分表的逻辑表名，如果是有很多需要水平分表的逻辑表，用逗号分隔

创建数据库SQL语句

SET NAMES utf8mb4;

SET FOREIGN_KEY_CHECKS = 0;

-- ----------------------------

-- Table structure for alarmhis201912

-- ----------------------------

DROP TABLE IF EXISTS `alarmhis201912`;

CREATE TABLE `alarmhis201912` (

`TurbineID` tinyint(0) UNSIGNED NOT NULL DEFAULT 0,

`AlmPointID` smallint(0) UNSIGNED NOT NULL,

`AlmHappenTime` datetime(0) NOT NULL,

`AlmClearTime` datetime(0) NULL DEFAULT NULL,

PRIMARY KEY (`TurbineID`, `AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `i_almID`(`AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `almTime`(`AlmHappenTime`) USING BTREE

) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

-- ----------------------------

-- Table structure for alarmhis202001

-- ----------------------------

DROP TABLE IF EXISTS `alarmhis202001`;

CREATE TABLE `alarmhis202001` (

`TurbineID` tinyint(0) UNSIGNED NOT NULL DEFAULT 0,

`AlmPointID` smallint(0) UNSIGNED NOT NULL,

`AlmHappenTime` datetime(0) NOT NULL,

`AlmClearTime` datetime(0) NULL DEFAULT NULL,

PRIMARY KEY (`TurbineID`, `AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `i_almID`(`AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `almTime`(`AlmHappenTime`) USING BTREE

) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

-- ----------------------------

-- Table structure for alarmhis202002

-- ----------------------------

DROP TABLE IF EXISTS `alarmhis202002`;

CREATE TABLE `alarmhis202002` (

`TurbineID` tinyint(0) UNSIGNED NOT NULL DEFAULT 0,

`AlmPointID` smallint(0) UNSIGNED NOT NULL,

`AlmHappenTime` datetime(0) NOT NULL,

`AlmClearTime` datetime(0) NULL DEFAULT NULL,

PRIMARY KEY (`TurbineID`, `AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `i_almID`(`AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `almTime`(`AlmHappenTime`) USING BTREE

) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

-- ----------------------------

-- Table structure for alarmhis202003

-- ----------------------------

DROP TABLE IF EXISTS `alarmhis202003`;

CREATE TABLE `alarmhis202003` (

`TurbineID` tinyint(0) UNSIGNED NOT NULL DEFAULT 0,

`AlmPointID` smallint(0) UNSIGNED NOT NULL,

`AlmHappenTime` datetime(0) NOT NULL,

`AlmClearTime` datetime(0) NULL DEFAULT NULL,

PRIMARY KEY (`TurbineID`, `AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `i_almID`(`AlmPointID`, `AlmHappenTime`) USING BTREE,

INDEX `almTime`(`AlmHappenTime`) USING BTREE

) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;