spring boot 八、 sharding-jdbc 分库分表 按月分表

在项目resources目录下新建com.jianmu.config.sharding.DateShardingAlgorithm 文件

新增yaml配置 数据源

spring:shardingsphere:props:sql:#是否在日志中打印 SQLshow: true#打印简单风格的 SQLsimple: truedatasource:names: pingxuanlogpingxuanlog:type: com.alibaba.druid.pool.DruidDataSourcedriver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://localhost:3306/db_jianmu_pingxuan_log?serverTimezone=UTC&useSSL=false&useUnicode=true&characterEncoding=UTF-8username: rootpassword: root#最大连接池数量max-active: 10#最小连接池数量min-idle: 5#初始化时建立物理连接的个数initial-size: 5#获取连接时最大等待时间，单位毫秒max-wait: 3000#配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒time-between-eviction-runs-millis: 60000#配置一个连接在池中最小生存的时间，单位是毫秒min-evictable-idle-time-millis: 100000#用来检测连接是否有效的sql，要求是一个查询语句validation-query: SELECT 1 FROM DUAL#建议配置为true，不影响性能，并且保证安全性。申请连接的时候检测，如果空闲时间大于timeBetweenEvictionRunsMillis，执行validationQuery检测连接是否有效。test-while-idle: true

新增yaml配置 sharding 分表规则

spring:shardingsphere:sharding:tables:t_act_vt_log:#配置数据节点，这里是按月分表，时间范围设置在202201 ~ 210012actual-data-nodes: pingxuanlog.t_act_vt_log_$->{202201..203012}table-strategy:standard:#使用标准分片策略，配置分片字段sharding-column: add_time# 精确匹配规则(自定义类)precise-algorithm-class-name: com.jianmu.config.sharding.DateShardingAlgorithm# 范围匹配规则(自定义类)range-algorithm-class-name: com.jianmu.config.sharding.DateShardingAlgorithmt_act_access_log:#配置数据节点，这里是按月分表，时间范围设置在202201 ~ 210012actual-data-nodes: pingxuanlog.t_act_access_log_$->{202201..203012}table-strategy:standard:#使用标准分片策略，配置分片字段sharding-column: add_time# 精确匹配规则(自定义类)precise-algorithm-class-name: com.jianmu.config.sharding.DateShardingAlgorithm# 范围匹配规则(自定义类)range-algorithm-class-name: com.jianmu.config.sharding.DateShardingAlgorithm

DataSourceConfiguration

package com.jianmu.config.sharding;import com.baomidou.dynamic.datasource.DynamicRoutingDataSource;
import com.baomidou.dynamic.datasource.provider.AbstractDataSourceProvider;
import com.baomidou.dynamic.datasource.provider.DynamicDataSourceProvider;
import com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DataSourceProperty;
import com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceAutoConfiguration;
import com.baomidou.dynamic.datasource.spring.boot.autoconfigure.DynamicDataSourceProperties;
import org.apache.shardingsphere.shardingjdbc.jdbc.core.datasource.ShardingDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Primary;import javax.sql.DataSource;
import java.util.Map;/*** 动态数据源配置：* <p>* 使用{@link com.baomidou.dynamic.datasource.annotation.DS}注解，切换数据源** <code>@DS(DataSourceConfiguration.SHARDING_DATA_SOURCE_NAME)</code>**/
@Configuration
@AutoConfigureBefore({DynamicDataSourceAutoConfiguration.class, SpringBootConfiguration.class})
public class DataSourceConfiguration {/*** 分表数据源名称*/public static final String SHARDING_DATA_SOURCE_NAME = "sharding";/*** 动态数据源配置项*/private final DynamicDataSourceProperties dynamicDataSourceProperties;private final ShardingDataSource shardingDataSource;@Autowiredpublic DataSourceConfiguration(DynamicDataSourceProperties dynamicDataSourceProperties, @Lazy ShardingDataSource shardingDataSource) {this.dynamicDataSourceProperties = dynamicDataSourceProperties;this.shardingDataSource = shardingDataSource;}/*** 将shardingDataSource放到了多数据源（dataSourceMap）中**/@Beanpublic DynamicDataSourceProvider dynamicDataSourceProvider() {Map<String, DataSourceProperty> datasourceMap = dynamicDataSourceProperties.getDatasource();return new AbstractDataSourceProvider() {@Overridepublic Map<String, DataSource> loadDataSources() {Map<String, DataSource> dataSourceMap = createDataSourceMap(datasourceMap);// 将 sharding jdbc 管理的数据源也交给动态数据源管理dataSourceMap.put(SHARDING_DATA_SOURCE_NAME, shardingDataSource);return dataSourceMap;}};}/*** 将动态数据源设置为首选的* 当spring存在多个数据源时, 自动注入的是首选的对象* 设置为主要的数据源之后，就可以支持sharding jdbc原生的配置方式*/@Primary@Beanpublic DataSource dataSource(DynamicDataSourceProvider dynamicDataSourceProvider) {DynamicRoutingDataSource dataSource = new DynamicRoutingDataSource();dataSource.setPrimary(dynamicDataSourceProperties.getPrimary());dataSource.setStrict(dynamicDataSourceProperties.getStrict());dataSource.setStrategy(dynamicDataSourceProperties.getStrategy());dataSource.setProvider(dynamicDataSourceProvider);dataSource.setP6spy(dynamicDataSourceProperties.getP6spy());dataSource.setSeata(dynamicDataSourceProperties.getSeata());return dataSource;}
}

package com.jianmu.config.sharding;import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.actuate.autoconfigure.jdbc.DataSourceHealthContributorAutoConfiguration;
import org.springframework.boot.actuate.health.AbstractHealthIndicator;
import org.springframework.boot.actuate.jdbc.DataSourceHealthIndicator;
import org.springframework.boot.jdbc.metadata.DataSourcePoolMetadataProvider;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;import javax.sql.DataSource;
import java.util.Map;/*** @author kong*/
@Configuration
public class DataSourceHealthConfig extends DataSourceHealthContributorAutoConfiguration {@Value("${spring.datasource.dbcp2.validation-query:select 1}")private String defaultQuery;public DataSourceHealthConfig(Map<String, DataSource> dataSources, ObjectProvider<DataSourcePoolMetadataProvider> metadataProviders) {super(dataSources, metadataProviders);}@Overrideprotected AbstractHealthIndicator createIndicator(DataSource source) {DataSourceHealthIndicator indicator = (DataSourceHealthIndicator) super.createIndicator(source);if (!StringUtils.hasText(indicator.getQuery())) {indicator.setQuery(defaultQuery);}return indicator;}
}

package com.jianmu.config.sharding;import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import com.jianmu.tools.ApplicationContextTools;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.shardingsphere.api.sharding.standard.PreciseShardingAlgorithm;
import org.apache.shardingsphere.api.sharding.standard.PreciseShardingValue;
import org.apache.shardingsphere.api.sharding.standard.RangeShardingAlgorithm;
import org.apache.shardingsphere.api.sharding.standard.RangeShardingValue;import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;/**sharding  分库分表策略配置：按照年月 如t_act_vt_log_202208**/
@Slf4j
public class DateShardingAlgorithm implements PreciseShardingAlgorithm<LocalDateTime>, RangeShardingAlgorithm<LocalDateTime> {private final DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyyMM");@Overridepublic String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<LocalDateTime> psv) {final String table = psv.getLogicTableName();final String prefix = table + "_";if (log.isDebugEnabled()) {log.debug("分表更新");}TableService tableService = (TableService) ApplicationContextTools.getBean(TableService.class);//添加日期final String logicTable = prefix + psv.getValue().format(this.format);boolean exist = tableService.exist(logicTable);if (log.isDebugEnabled()) {log.debug("更新: {} 存在:{}", logicTable, exist);}//匹配到了if (exist) {return logicTable;}//未匹配到if (log.isDebugEnabled()) {log.debug("sharding 未匹配到表，需要创建");}//创建这张表tableService.copy(logicTable, table);return logicTable;}/*** 范围匹配*/@SneakyThrows@Overridepublic Collection<String> doSharding(Collection<String> collection, RangeShardingValue<LocalDateTime> rsv) {final String table = rsv.getLogicTableName();final String prefix = table + "_";if (log.isDebugEnabled()) {log.debug("分表查询 表:{}", table);}TableService tableService = (TableService) ApplicationContextTools.getBean(TableService.class);List<String> tables = tableService.tables();Range<LocalDateTime> range = rsv.getValueRange();//计算出时间范围内的所有日期final int upper = range.hasUpperBound() ? Integer.parseInt(range.upperEndpoint().toLocalDate().format(this.format)) : 0;final int lower = range.hasLowerBound() ? Integer.parseInt(range.lowerEndpoint().toLocalDate().format(this.format)) : 0;List<String> validTables = this.validTables(prefix, tables, upper, lower);if (log.isDebugEnabled()) {log.debug("查询表:{}", JSON.toJSONString(validTables));}if (CollectionUtils.isEmpty(validTables)) {if (log.isDebugEnabled()) {log.debug("查询表不存在 改查原始表");}return Lists.newArrayList(table);}return validTables;}private List<String> validTables(String prefix, List<String> tables, int upper, int lower) {if (log.isDebugEnabled()) {log.debug("上界:{},下界:{}", upper, lower);}return tables.parallelStream().filter(i -> {if (i.contains(prefix)) {final String date = i.replace(prefix, "");if (date.matches("[0-9]*")) {final int mouth = Integer.parseInt(date);if (upper > 0 && lower > 0) {return mouth <= upper && mouth >= lower;} else {if (upper > 0) {return mouth <= upper;}if (lower > 0) {return mouth >= lower;}}}}return false;}).collect(Collectors.toList());}}

package com.jianmu.config.sharding;public class ShardingConstant {public static final String ORIGINAL_DATABASE = "db_jianmu_pingxuan_log";
}

package com.jianmu.config.sharding;import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.ExecutionException;public interface TableService {/*** 返回所有表*/List<String> tables() throws ExecutionException, SQLException;boolean exist(String table);/*** 精确删除表*/boolean drop(String table);/*** 旧表复制为新表*/boolean copy(String newTable, String usedTable);
}

package com.jianmu.config.sharding;import com.baomidou.dynamic.datasource.DynamicRoutingDataSource;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.jianmu.constant.SysRedisConstant;
import com.jianmu.mapper.system.TableMapper;
import com.jianmu.tools.JdbcTools;
import com.jianmu.tools.log.LogTools;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;import java.sql.SQLException;
import java.util.List;@Slf4j
@Service
@DS("sharding")
public class TableServiceImpl implements TableService {private final TableMapper tableMapper;private final RedisTemplate<String, Object> redis;private final ValueOperations<String, Object> redisValue;private final DynamicRoutingDataSource dynamicRoutingDataSource;@Autowiredpublic TableServiceImpl(TableMapper tableMapper, RedisTemplate<String, Object> redis, DynamicRoutingDataSource dynamicRoutingDataSource) {this.tableMapper = tableMapper;this.redis = redis;this.redisValue = redis.opsForValue();this.dynamicRoutingDataSource = dynamicRoutingDataSource;}@Overridepublic List<String> tables() throws SQLException {List<String> tables = (List<String>) this.redisValue.get(SysRedisConstant.TABLES);if (tables == null) {List<String> list = JdbcTools.tables(this.dynamicRoutingDataSource.getDataSource(DataSourceConfiguration.SHARDING_DATA_SOURCE_NAME).getConnection(), ShardingConstant.ORIGINAL_DATABASE);//每24小时更新一次this.redisValue.set(SysRedisConstant.TABLES, list, 86400);return list;}return tables;}@Overridepublic boolean exist(String table) {try {Integer exist = this.tableMapper.exist(table);if (log.isDebugEnabled()) {log.debug("表:{} 存在:{}", table, exist);}return true;} catch (Exception e) {LogTools.err(e);return false;}}@Overridepublic boolean drop(String table) {boolean flag = this.tableMapper.drop(table) >= 0;if (flag) {this.delCache();}return flag;}@Overridepublic boolean copy(String newTable, String usedTable) {boolean flag = this.tableMapper.copy(newTable, usedTable) >= 0;if (flag) {this.delCache();}return flag;}private void delCache() {this.redis.delete(SysRedisConstant.TABLES);}}

t_act_vt_log 为每次自动生成的基础表 需要在这个表建好索引等