shardingsphere 5.5.1按月分表配置记录，支持dynamic-datasource多数据源

未兆

2285人浏览 · 2024-11-11 16:22:06

未兆 · 2024-11-11 16:22:06 发布

新版本的shardingsphere配置查了很多材料，最后终于配置完成，做个记录。springboot版本2.7.18

先说明下springboot的支持只到5.2.1，但是因为旧版本有漏洞问题，加上官方新版本不在支持springboot写法，所以改成了shardingsphere-jdbc。

maven的版本

<shardingsphere.version>5.5.1</shardingsphere.version>
<snakeyaml.version>2.3</snakeyaml.version>

<dependency>
    <groupId>org.apache.shardingsphere</groupId>
    <artifactId>shardingsphere-jdbc</artifactId>
    <version>${shardingsphere.version}</version>
</dependency>
<dependency>
    <groupId>org.yaml</groupId>
    <artifactId>snakeyaml</artifactId>
    <version>${snakeyaml.version}</version>
</dependency>

sharding.xml

dataSources:
  nemesismaster:
    dataSourceClassName:  com.alibaba.druid.pool.DruidDataSource
    driverClassName: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://xxx/nemesis?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&zeroDateTimeBehavior=CONVERT_TO_NULL&autoReconnect=true&failOverReadOnly=false&serverTimezone=GMT%2B8&useSSL=false
    username: xxx
    password: xxx

rules:
  - !SHARDING
    tables:
      risk_record:
        actualDataNodes: nemesismaster.risk_record
        tableStrategy:
          standard:
            shardingColumn: create_time
            shardingAlgorithmName: aaa

    shardingAlgorithms:
      aaa:
        type: CLASS_BASED
        props:
          strategy: standard
          algorithmClassName: com.xxx.xxx.nemesis.shardingsphere.TimeShardingAlgorithm
#      interval:
#        type: INTERVAL
#        props:
#          datetime-pattern: 'yyyy-MM'
#          datetime-lower: '2024-01'
#          datetime-upper: '2025-12'
#          sharding-suffix-pattern: 'yyyyMM'
#          # 间隔大小
#          datetime-interval-amount: 1
#          datetime-interval-unit: 'Months'
  - !SINGLE
    tables:
      - "*.*"
props:
  sql-show: true

yml的配置里面加上

 spring:
  datasource:
    dynamic:
      datasource:
        shardingSphere:
          url: jdbc:shardingsphere:classpath:sharding.yml
          driver-class-name: org.apache.shardingsphere.driver.ShardingSphereDriver

分片算法

package com.xxx.xxx.nemesis.shardingsphere;

import com.google.common.collect.Range;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.sharding.api.sharding.ShardingAutoTableAlgorithm;
import org.apache.shardingsphere.sharding.api.sharding.standard.PreciseShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.RangeShardingValue;
import org.apache.shardingsphere.sharding.api.sharding.standard.StandardShardingAlgorithm;

import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 分片算法，按月分片
 */
@Slf4j
public final class TimeShardingAlgorithm implements StandardShardingAlgorithm<Timestamp>, ShardingAutoTableAlgorithm {

    /**
     * 分片时间格式
     */
    private static final DateTimeFormatter TABLE_SHARD_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMM");

    /**
     * 完整时间格式
     */
    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd HH:mm:ss");

    /**
     * 表分片符号，例：t_contract_202201 中，分片符号为 "_"
     */
    private final String TABLE_SPLIT_SYMBOL = "_";

    @Getter
    private Properties props;

    @Getter
    private int autoTablesAmount;

    @Override
    public void init(final Properties props) {
        this.props = props;
    }

    public LocalDateTime timestampToDatetime(long timestamp) {
        Instant instant = Instant.ofEpochMilli(timestamp);
        return LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
    }

    @Override
    public String doSharding(final Collection<String> availableTargetNames, final PreciseShardingValue<Timestamp> preciseShardingValue) {
        String logicTableName = preciseShardingValue.getLogicTableName();

        /// 打印分片信息
        log.info(">>>>>>>>>> 【INFO】精确分片，节点配置表名：{}", availableTargetNames);

        LocalDateTime dateTime = preciseShardingValue.getValue().toLocalDateTime();
        String resultTableName = logicTableName + "_" + dateTime.format(TABLE_SHARD_TIME_FORMATTER);

        // 检查是否需要初始化
        if (availableTargetNames.size() == 1) {
            // 如果只有一个表，说明需要获取所有表名
            List<String> allTableNameBySchema = ShardingAlgorithmTool.getAllTableNameBySchema(logicTableName);
            availableTargetNames.clear();
            availableTargetNames.addAll(allTableNameBySchema);
            autoTablesAmount = allTableNameBySchema.size();
            //没有找到则创建一个
            if (autoTablesAmount == 0) {
                return getShardingTableAndCreate(logicTableName, resultTableName, availableTargetNames);
            }
            return resultTableName;
        }

        return getShardingTableAndCreate(logicTableName, resultTableName, availableTargetNames);
    }

    @Override
    public Collection<String> doSharding(final Collection<String> availableTargetNames, final RangeShardingValue<Timestamp> rangeShardingValue) {
        String logicTableName = rangeShardingValue.getLogicTableName();

        /// 打印分片信息
        log.info(">>>>>>>>>> 【INFO】范围分片，节点配置表名：{}", availableTargetNames);

        // between and 的起始值
        Range<Timestamp> valueRange = rangeShardingValue.getValueRange();
        boolean hasLowerBound = valueRange.hasLowerBound();
        boolean hasUpperBound = valueRange.hasUpperBound();

        // 获取最大值和最小值
        LocalDateTime min = hasLowerBound ? valueRange.lowerEndpoint().toLocalDateTime() : getLowerEndpoint(availableTargetNames);
        LocalDateTime max = hasUpperBound ? valueRange.upperEndpoint().toLocalDateTime() : getUpperEndpoint(availableTargetNames);

        // 循环计算分表范围
        Set<String> resultTableNames = new LinkedHashSet<>();
        while (min.isBefore(max) || min.equals(max)) {
            String tableName = logicTableName + TABLE_SPLIT_SYMBOL + min.format(TABLE_SHARD_TIME_FORMATTER);
            resultTableNames.add(tableName);
            min = min.plusMinutes(1);
        }
        return getShardingTablesAndCreate(logicTableName, resultTableNames, availableTargetNames);
    }

    @Override
    public String getType() {
        return "AUTO_CUSTOM";
    }

    /**
     * 检查分表获取的表名是否存在，不存在则自动建表
     *
     * @param logicTableName       逻辑表
     * @param resultTableNames     真实表名，例：t_user_202201
     * @param availableTargetNames 可用的数据库表名
     * @return 存在于数据库中的真实表名集合
     */
    public Set<String> getShardingTablesAndCreate(String logicTableName, Collection<String> resultTableNames, Collection<String> availableTargetNames) {
        return resultTableNames.stream().map(o -> getShardingTableAndCreate(logicTableName, o, availableTargetNames)).collect(Collectors.toSet());
    }

    /**
     * 检查分表获取的表名是否存在，不存在则自动建表
     *
     * @param logicTableName  逻辑表
     * @param resultTableName 真实表名，例：t_user_202201
     * @return 确认存在于数据库中的真实表名
     */
    private String getShardingTableAndCreate(String logicTableName, String resultTableName, Collection<String> availableTargetNames) {
        // 缓存中有此表则返回，没有则判断创建
        if (availableTargetNames.contains(resultTableName)) {
            return resultTableName;
        } else {
            // 检查分表获取的表名不存在，需要自动建表
            boolean isSuccess = ShardingAlgorithmTool.createShardingTable(logicTableName, resultTableName);
            if (isSuccess) {
                // 如果建表成功，需要更新缓存
                availableTargetNames.add(resultTableName);
                autoTablesAmount++;
                return resultTableName;
            } else {
                // 如果建表失败，返回逻辑空表
                return logicTableName;
            }
        }
    }

    /**
     * 获取 最小分片值
     *
     * @param tableNames 表名集合
     * @return 最小分片值
     */
    private LocalDateTime getLowerEndpoint(Collection<String> tableNames) {
        Optional<LocalDateTime> optional = tableNames.stream()
                .map(o -> LocalDateTime.parse(o.replace(TABLE_SPLIT_SYMBOL, "") + "01 00:00:00", DATE_TIME_FORMATTER))
                .min(Comparator.comparing(Function.identity()));
        if (optional.isPresent()) {
            return optional.get();
        } else {
            log.error(">>>>>>>>>> 【ERROR】获取数据最小分表失败，请稍后重试，tableName：{}", tableNames);
            throw new IllegalArgumentException("获取数据最小分表失败，请稍后重试");
        }
    }

    /**
     * 获取 最大分片值
     *
     * @param tableNames 表名集合
     * @return 最大分片值
     */
    private LocalDateTime getUpperEndpoint(Collection<String> tableNames) {
        Optional<LocalDateTime> optional = tableNames.stream()
                .map(o -> LocalDateTime.parse(o.replace(TABLE_SPLIT_SYMBOL, "") + "01 00:00:00", DATE_TIME_FORMATTER))
                .max(Comparator.comparing(Function.identity()));
        if (optional.isPresent()) {
            return optional.get();
        } else {
            log.error(">>>>>>>>>> 【ERROR】获取数据最大分表失败，请稍后重试，tableName：{}", tableNames);
            throw new IllegalArgumentException("获取数据最大分表失败，请稍后重试");
        }
    }
}

按月分片算法工具

package com.xxx.xxx.nemesis.shardingsphere;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.core.env.Environment;

import java.sql.*;
import java.time.YearMonth;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 按月分片算法工具
 */
@Slf4j
public class ShardingAlgorithmTool {

    /**
     * 表分片符号，例：t_user_202201 中，分片符号为 "_"
     */
    private static final String TABLE_SPLIT_SYMBOL = "_";

    /**
     * 数据库配置
     */
    private static final Environment ENV = SpringUtil.getApplicationContext().getEnvironment();
    private static final String DATASOURCE_URL = ENV.getProperty("my.sharding.create-table.url");
    private static final String DATASOURCE_USERNAME = ENV.getProperty("my.sharding.create-table.username");
    private static final String DATASOURCE_PASSWORD = ENV.getProperty("my.sharding.create-table.password");

    /**
     * 获取所有表名
     *
     * @param logicTableName 逻辑表
     * @return 表名集合
     */
    public static List<String> getAllTableNameBySchema(String logicTableName) {
        List<String> tableNames = new ArrayList<>();
        if (StringUtils.isEmpty(DATASOURCE_URL) || StringUtils.isEmpty(DATASOURCE_USERNAME) || StringUtils.isEmpty(DATASOURCE_PASSWORD)) {
            log.error(">>>>>>>>>> 【ERROR】数据库连接配置有误，请稍后重试，URL:{}, username:{}, password:{}", DATASOURCE_URL, DATASOURCE_USERNAME, DATASOURCE_PASSWORD);
            throw new IllegalArgumentException("数据库连接配置有误，请稍后重试");
        }
        try (Connection conn = DriverManager.getConnection(DATASOURCE_URL, DATASOURCE_USERNAME, DATASOURCE_PASSWORD);
             Statement st = conn.createStatement()) {
            try (ResultSet rs = st.executeQuery("show TABLES like '" + logicTableName + TABLE_SPLIT_SYMBOL + "%'")) {
                while (rs.next()) {
                    String tableName = rs.getString(1);
                    // 匹配分表格式 例：^(t\_contract_\d{6})$
                    if (tableName != null && tableName.matches(String.format("^(%s\\d{6})$", logicTableName + TABLE_SPLIT_SYMBOL))) {
                        tableNames.add(rs.getString(1));
                    }
                }
            }
        } catch (SQLException e) {
            log.error(">>>>>>>>>> 【ERROR】数据库连接失败，请稍后重试，原因：{}", e.getMessage(), e);
            throw new IllegalArgumentException("数据库连接失败，请稍后重试");
        }
        return tableNames;
    }

    /**
     * 创建分表2
     *
     * @param logicTableName  逻辑表
     * @param resultTableName 真实表名，例：t_user_202201
     * @return 创建结果（true创建成功，false未创建）
     */
    public static boolean createShardingTable(String logicTableName, String resultTableName) {
        // 根据日期判断，当前月份之后分表不提前创建
        String month = resultTableName.replace(logicTableName + TABLE_SPLIT_SYMBOL, "");
        YearMonth shardingMonth = YearMonth.parse(month, DateTimeFormatter.ofPattern("yyyyMM"));
        if (shardingMonth.isAfter(YearMonth.now())) {
            return false;
        }

        synchronized (logicTableName.intern()) {
            // 缓存中无此表，则建表并添加缓存
            executeSql(Collections.singletonList("CREATE TABLE IF NOT EXISTS `" + resultTableName + "` LIKE `" + logicTableName + "`;"));
        }
        return true;
    }

    /**
     * 执行SQL
     *
     * @param sqlList SQL集合
     */
    private static void executeSql(List<String> sqlList) {
        if (StringUtils.isEmpty(DATASOURCE_URL) || StringUtils.isEmpty(DATASOURCE_USERNAME) || StringUtils.isEmpty(DATASOURCE_PASSWORD)) {
            log.error(">>>>>>>>>> 【ERROR】数据库连接配置有误，请稍后重试，URL:{}, username:{}, password:{}", DATASOURCE_URL, DATASOURCE_USERNAME, DATASOURCE_PASSWORD);
            throw new IllegalArgumentException("数据库连接配置有误，请稍后重试");
        }
        try (Connection conn = DriverManager.getConnection(DATASOURCE_URL, DATASOURCE_USERNAME, DATASOURCE_PASSWORD)) {
            try (Statement st = conn.createStatement()) {
                conn.setAutoCommit(false);
                for (String sql : sqlList) {
                    st.execute(sql);
                }
            } catch (Exception e) {
                conn.rollback();
                log.error(">>>>>>>>>> 【ERROR】数据表创建执行失败，请稍后重试，原因：{}", e.getMessage(), e);
                throw new IllegalArgumentException("数据表创建执行失败，请稍后重试");
            }
        } catch (SQLException e) {
            log.error(">>>>>>>>>> 【ERROR】数据库连接失败，请稍后重试，原因：{}", e.getMessage(), e);
            throw new IllegalArgumentException("数据库连接失败，请稍后重试");
        }
    }
}