devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Springboot+shardingsphere实现一库多表分表(对多个表进行分表操作)_shardingspheredatasourcefactory.createdatasource

作者：从前慢现在也慢 | 2024-06-25 19:02:23

赞

踩

shardingspheredatasourcefactory.createdatasource

        Springboot+shardingsphere实现多一个数据库里面的多个表进行分表操作，本文是针对mysql数据库中，两个表进行分表操作，实现根据分表字段自动创建不存在的表以及自动根据分表字段实现查询数据库操作。所使用的主要技术栈为shardingsphere+sqi。

maven依赖

  1.         <dependency>
  2. 			<groupId>org.apache.shardingsphere</groupId>
  3. 			<artifactId>shardingsphere-jdbc-core</artifactId>
  4. 			<version>5.0.0</version>
  5. 		</dependency>
  6.  
  7. 		<dependency>
  8. 			<groupId>velocity</groupId>
  9. 			<artifactId>velocity</artifactId>
  10. 			<version>1.4</version>
  11. 		</dependency>

config文件

  1. @Slf4j
  2. @Configuration
  3. public class ShardingJdbcConfig {
  4.  
  5.   private final DataSourceProperties dataSourceProperties;
  6.  
  7.   public  static HikariDataSource hikariDataSource;
  8.  
  9.   static {
  10.     Velocity.setProperty("resource.loader", "class");
  11.     Velocity.setProperty("class.resource.loader.class",
  12.         ClasspathResourceLoader.class.getName());
  13.     try {
  14.       Velocity.init();
  15.     } catch (Exception e) {
  16.       e.printStackTrace();
  17.     }
  18.   }
  19.  
  20.   public ShardingJdbcConfig(DataSourceProperties dataSourceProperties) {
  21.     this.dataSourceProperties = dataSourceProperties;
  22.   }
  23.  
  24.   public HikariDataSource jdbcDatasource() {
  25.     HikariDataSource dataSource = new HikariDataSource();
  26.     BeanUtils.copyProperties(dataSourceProperties,dataSource);
  27.     dataSource.setDriverClassName(dataSourceProperties.getDriverClassName());
  28.     dataSource.setJdbcUrl(dataSourceProperties.getUrl());
  29.     dataSource.setUsername(dataSourceProperties.getUsername());
  30.     dataSource.setPassword(dataSourceProperties.getPassword());
  31.     dataSource.setConnectionTimeout(1000 * 60);
  32.     hikariDataSource = dataSource;
  33.     return dataSource;
  34.   }
  35.  
  36.   @Bean
  37.   @Primary
  38.   @Qualifier("ds")
  39.   public DataSource dataSource() throws SQLException {
  40.     Map<String, DataSource> dataSourceMap = new HashMap<>();
  41.     HikariDataSource dataSource = new HikariDataSource();
  42.     dataSource.setDriverClassName(dataSourceProperties.getDriverClassName());
  43.     dataSource.setJdbcUrl(dataSourceProperties.getUrl());
  44.     dataSource.setUsername(dataSourceProperties.getUsername());
  45.     dataSource.setPassword(dataSourceProperties.getPassword());
  46.     dataSourceMap.put("ds", dataSource);
  47.  
  48.     // 配置 record 表规则
  49.     ShardingTableRuleConfiguration recordTableRuleConfiguration = new ShardingTableRuleConfiguration("t_test_table", getActualDataNodes());
  50.  
  51.     // 配置分表策略
  52.     recordTableRuleConfiguration.setTableShardingStrategy(new StandardShardingStrategyConfiguration("create_time", KafkaTestTableStandardShardingAlgorithm.TYPE));
  53.  
  54.     // 配置分片规则
  55.     ShardingRuleConfiguration shardingRuleConfig = new ShardingRuleConfiguration();
  56.     shardingRuleConfig.getTables().add(recordTableRuleConfiguration);
  57.  
  58.     // 配置分表算法
  59.     Properties tableShardingAlgorithmProps = new Properties();
  60.     tableShardingAlgorithmProps.put("jdbcDatasource", jdbcDatasource());
  61.     shardingRuleConfig.getShardingAlgorithms().put(KafkaTestTableStandardShardingAlgorithm.TYPE,
  62.         new ShardingSphereAlgorithmConfiguration(KafkaTestTableStandardShardingAlgorithm.TYPE, tableShardingAlgorithmProps)
  63.     );
  64.  
  65.     // 配置 t_redp_face_match_event 表规则
  66.     ShardingTableRuleConfiguration recordTableRuleConfigurationRedp = new ShardingTableRuleConfiguration("t_kafka_test", getActualDataNodesRedp());
  67.     // 配置分表策略
  68.     recordTableRuleConfigurationRedp.setTableShardingStrategy(new StandardShardingStrategyConfiguration("create_time", KafkaTestTableStandardShardingAlgorithm.TYPE)
  69.     );
  70.  
  71.     // 配置分片规则
  72.     shardingRuleConfig.getTables().add(recordTableRuleConfigurationRedp);
  73.  
  74.     // 配置分表算法
  75.     shardingRuleConfig.getShardingAlgorithms().put(KafkaTestTableStandardShardingAlgorithm.TYPE, new ShardingSphereAlgorithmConfiguration(KafkaTestTableStandardShardingAlgorithm.TYPE, tableShardingAlgorithmProps));
  76.  
  77.  
  78.     List listConfig = new ArrayList<ShardingRuleConfiguration>(){
  79.       {
  80.         add(shardingRuleConfig);
  81.       }
  82.     };
  83.     // 创建 ShardingSphereDataSource
  84.     return ShardingSphereDataSourceFactory.createDataSource(dataSourceMap, listConfig, tableShardingAlgorithmProps);
  85.   }
  86.  
  87.   private String getActualDataNodes() {
  88.       return "ds.t_test_table${2021..2099}0${1..9},ds.t_test_table${2021..2099}1${0..2}";
  89.     }
  90.  
  91.   private String getActualDataNodesRedp() {
  92.     return "ds.t_kafka_test${2021..2099}0${1..9},ds.t_kafka_test${2021..2099}1${0..2}";
  93.   }
  94.  
  95. }

两个表的分表算法逻辑

  1. @Slf4j
  2. public class KafkaTestTableStandardShardingAlgorithm implements StandardShardingAlgorithm<Date> {
  3.  
  4.     public static final String TYPE = "KafkaTestTableStandardShardingAlgorithm";
  5.  
  6.     private List<DataSource> dataSources;
  7.  
  8.     private final Set<String> TABLES = new ConcurrentHashSet<>();
  9.  
  10.     protected Properties props;
  11.     protected String initDate;
  12.  
  13.     private  static ThreadLocal<DateFormat> dateformat = new ThreadLocal<>();
  14.  
  15.     @Override
  16.     public String doSharding(Collection<String> collection, PreciseShardingValue<Date> shardingValue) {
  17.         StringBuilder tableName = new StringBuilder();
  18.         Date value = shardingValue.getValue();
  19.         tableName.append(shardingValue.getLogicTableName()).append(getDateFormatter().format(value));
  20.         log.info("执行操作的表名{}",tableName.toString());
  21.         if (!TABLES.contains(tableName.toString())) {
  22.             createTable(tableName.toString());
  23.         }
  24.         return tableName.toString();
  25.     }
  26.  
  27.     @Override
  28.     public Collection<String> doSharding(Collection<String> collection, RangeShardingValue<Date> rangeShardingValue) {
  29.         Collection<String> result = new LinkedHashSet<>();
  30.  
  31.         Range<Date> shardingKey = rangeShardingValue.getValueRange();
  32.  
  33.         Date startTime = shardingKey.lowerEndpoint();
  34.  
  35.         Date endTime = shardingKey.upperEndpoint();
  36.  
  37.         // 获取起始，终止时间范围
  38.         Date now = new Date();
  39.         if (startTime.after(now)) {
  40.             startTime = now;
  41.         }
  42.         if (endTime.after(now)) {
  43.             endTime = now;
  44.         }
  45.         Collection<String> tables = getRoutTable(rangeShardingValue.getLogicTableName(), startTime, endTime);
  46.  
  47.         if (tables != null && tables.size() > 0) {
  48.             result.addAll(tables);
  49.         }
  50.         return result;
  51.     }
  52.  
  53.     @Override
  54.     public void init() {
  55.         this.dataSources = new ArrayList<DataSource>(){
  56.             {
  57.                 add(ShardingJdbcConfig.hikariDataSource);
  58.             }
  59.         };
  60.         this.syncTable();
  61.     }
  62.  
  63.     @Override
  64.     public String getType() {
  65.         return TYPE;
  66.     }
  67.  
  68.     @Override
  69.     public Properties getProps() {
  70.         return props;
  71.     }
  72.  
  73.     @Override
  74.     public void setProps(Properties props) {
  75.         this.props = props;
  76.     }
  77.  
  78.     private void syncTable() {
  79.         String sql = "select TABLE_NAME from information_schema.TABLES " +
  80.                 "where TABLE_NAME like 't_kafka_test%'";
  81.         for (DataSource dataSource : this.dataSources) {
  82.             try (PreparedStatement preparedStatement = dataSource.getConnection().prepareStatement(sql);
  83.                  ResultSet rs = preparedStatement.executeQuery()) {
  84.                 while (rs.next()) {
  85.                     TABLES.add(rs.getString(1));
  86.                 }
  87.             } catch (SQLException e) {
  88.                 log.error("sync table error:{}", e.getMessage(), e);
  89.             }
  90.         }
  91.     }
  92.  
  93.     private void createTable(String tableName) {
  94.         try {
  95.             Template template =
  96.                     Velocity.getTemplate("template/ddlKafkaTest.template");
  97.             VelocityContext context = new VelocityContext();
  98.             context.put("tableName", tableName);
  99.             StringWriter sw = new StringWriter();
  100.             template.merge(context, sw);
  101.             Resource resource = new ByteArrayResource(sw.toString().getBytes(StandardCharsets.UTF_8));
  102.             for (DataSource dataSource : this.dataSources) {
  103.                 try (Connection connection = dataSource.getConnection()) {
  104.                     ScriptUtils.executeSqlScript(connection, resource);
  105.                 }
  106.             }
  107.             TABLES.add(tableName);
  108.         } catch (Exception e) {
  109.             log.error("create table({}) error:{}", tableName, e.getMessage(), e);
  110.         }
  111.     }
  112.  
  113.     private static DateFormat getDateFormatter(){
  114.         DateFormat dateFormat = dateformat.get();
  115.         if (dateFormat == null) {
  116.             dateFormat = new SimpleDateFormat("yyyyMM");
  117.             dateformat.set(dateFormat);
  118.         }
  119.         return dateFormat;
  120.     }
  121.  
  122.     private Collection<String> getRoutTable(String logicTableName, Date startTime, Date endTime) {
  123.         Set<String> rouTables = new HashSet<>();
  124.         if (startTime != null && endTime != null) {
  125.             List<String> rangeNameList = getRangeNameList(startTime, endTime);
  126.             for (String YearMonth : rangeNameList) {
  127.                 String tableName = logicTableName + YearMonth;
  128.                 if (!TABLES.contains(tableName)) {
  129.                     createTable(tableName);
  130.                 }
  131.                 rouTables.add(tableName);
  132.             }
  133.         }
  134.         return rouTables;
  135.     }
  136.  
  137.     private static List<String> getRangeNameList(Date startTime, Date endTime) {
  138.         List<String> result = Lists.newArrayList();
  139.  
  140.         DateTime beginOfMonth = cn.hutool.core.date.DateUtil.beginOfMonth(startTime);
  141.         DateTime endOfMonth = cn.hutool.core.date.DateUtil.beginOfMonth(endTime);
  142.  
  143.  
  144.         while (beginOfMonth.getTime() <= endOfMonth.getTime()) {
  145.             result.add(getDateFormatter().format(beginOfMonth.getTime()));
  146.             // 进行当前日期按月份 + 1
  147.             beginOfMonth = beginOfMonth.offset(DateField.MONTH, 1);
  148.         }
  149.         return result;
  150.     }
  151. }
  1. @Slf4j
  2. public class TestTableStandardShardingAlgorithm implements StandardShardingAlgorithm<Date> {
  3.  
  4.     public static final String TYPE = "TestTableStandardShardingAlgorithm";
  5.  
  6.     private List<DataSource> dataSources;
  7.  
  8.     private final Set<String> TABLES = new ConcurrentHashSet<>();
  9.  
  10.     protected Properties props;
  11.     protected String initDate;
  12.  
  13.     private  static ThreadLocal<DateFormat> dateformat = new ThreadLocal<>();
  14.  
  15.     @Override
  16.     public String doSharding(Collection<String> collection, PreciseShardingValue<Date> shardingValue) {
  17.         StringBuilder tableName = new StringBuilder();
  18.         Date startTime = shardingValue.getValue();
  19.  
  20.         log.info("执行操作的表名{}", shardingValue.getLogicTableName() + getDateFormatter().format(startTime));
  21.         tableName.append(shardingValue.getLogicTableName()).append(getDateFormatter().format(startTime));
  22.  
  23.         if (!TABLES.contains(tableName.toString())) {
  24.             createTable(tableName.toString());
  25.         }
  26.         return tableName.toString();
  27.     }
  28.  
  29.     @Override
  30.     public Collection<String> doSharding(Collection<String> collection, RangeShardingValue<Date> rangeShardingValue) {
  31.         Collection<String> result = new LinkedHashSet<>();
  32.  
  33.         Range<Date> shardingKey = rangeShardingValue.getValueRange();
  34.  
  35.         Date startTime = shardingKey.lowerEndpoint();
  36.  
  37.         Date endTime = shardingKey.upperEndpoint();
  38.  
  39.         // 获取起始，终止时间范围
  40.         Date now = new Date();
  41.         if (startTime.after(now)) {
  42.             startTime = now;
  43.         }
  44.         if (endTime.after(now)) {
  45.             endTime = now;
  46.         }
  47.         Collection<String> tables = getRoutTable(rangeShardingValue.getLogicTableName(), startTime, endTime);
  48.  
  49.         if (tables != null && tables.size() > 0) {
  50.             result.addAll(tables);
  51.         }
  52.         return result;
  53.     }
  54.  
  55.     @Override
  56.     public void init() {
  57.         this.dataSources = new ArrayList<DataSource>(){
  58.             {
  59.                 add(ShardingJdbcConfig.hikariDataSource);
  60.             }
  61.         };
  62.         this.syncTable();
  63.     }
  64.  
  65.     @Override
  66.     public String getType() {
  67.         return TYPE;
  68.     }
  69.  
  70.     @Override
  71.     public Properties getProps() {
  72.         return props;
  73.     }
  74.  
  75.     @Override
  76.     public void setProps(Properties props) {
  77.         this.props = props;
  78.     }
  79.  
  80.     private void syncTable() {
  81.         String sql = "select TABLE_NAME from information_schema.TABLES " +
  82.                 "where TABLE_NAME like 't_test_table%'";
  83.         for (DataSource dataSource : this.dataSources) {
  84.             try (PreparedStatement preparedStatement =
  85.                          dataSource.getConnection().prepareStatement(sql);
  86.                  ResultSet rs = preparedStatement.executeQuery()) {
  87.                 while (rs.next()) {
  88.                     TABLES.add(rs.getString(1));
  89.                 }
  90.             } catch (SQLException e) {
  91.                 log.error("sync table error:{}", e.getMessage(), e);
  92.             }
  93.         }
  94.     }
  95.  
  96.     private void createTable(String tableName) {
  97.         try {
  98.             Template template =
  99.                     Velocity.getTemplate("template/ddl.template");
  100.             VelocityContext context = new VelocityContext();
  101.             context.put("tableName", tableName);
  102.             StringWriter sw = new StringWriter();
  103.             template.merge(context, sw);
  104.             Resource resource =
  105.                     new ByteArrayResource(sw.toString().getBytes(StandardCharsets.UTF_8));
  106.             for (DataSource dataSource : this.dataSources) {
  107.                 try (Connection connection = dataSource.getConnection()) {
  108.                     ScriptUtils.executeSqlScript(connection, resource);
  109.                 }
  110.             }
  111.             TABLES.add(tableName);
  112.         } catch (Exception e) {
  113.             log.error("create table({}) error:{}", tableName, e.getMessage(), e);
  114.         }
  115.     }
  116.  
  117.     private static DateFormat getDateFormatter(){
  118.         DateFormat dateFormat = dateformat.get();
  119.         if (dateFormat == null) {
  120.             dateFormat = new SimpleDateFormat("yyyyMM");
  121.             dateformat.set(dateFormat);
  122.         }
  123.         return dateFormat;
  124.     }
  125.  
  126.     private Collection<String> getRoutTable(String logicTableName, Date startTime, Date endTime) {
  127.         Set<String> rouTables = new HashSet<>();
  128.         if (startTime != null && endTime != null) {
  129.             List<String> rangeNameList = getRangeNameList(startTime, endTime);
  130.             for (String YearMonth : rangeNameList) {
  131.                 String tableName = logicTableName + YearMonth;
  132.                 if (!TABLES.contains(tableName)) {
  133.                     createTable(tableName);
  134.                 }
  135.                 rouTables.add(tableName);
  136.             }
  137.         }
  138.         return rouTables;
  139.     }
  140.  
  141.     private static List<String> getRangeNameList(Date startTime, Date endTime) {
  142.         List<String> result = Lists.newArrayList();
  143.  
  144.         DateTime beginOfMonth = cn.hutool.core.date.DateUtil.beginOfMonth(startTime);
  145.         DateTime endOfMonth = cn.hutool.core.date.DateUtil.beginOfMonth(endTime);
  146.  
  147.  
  148.         while (beginOfMonth.getTime() <= endOfMonth.getTime()) {
  149.             result.add(getDateFormatter().format(beginOfMonth.getTime()));
  150.             // 进行当前日期按月份 + 1
  151.             beginOfMonth = beginOfMonth.offset(DateField.MONTH, 1);
  152.         }
  153.         return result;
  154.     }
  155. }

数据库表模板文件

模板文件目录如下

 模板sql

  1. CREATE TABLE if not exists ${tableName} (
  2.   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  3.   `order_id` varchar(200) NOT NULL,
  4.   `name` varchar(255) DEFAULT NULL COMMENT '收货人',
  5.   `address` varchar(255) DEFAULT NULL COMMENT '收货人地址',
  6.   `phone` varchar(255) DEFAULT NULL COMMENT '联系电话',
  7.   `create_time` datetime DEFAULT NULL,
  8.   PRIMARY KEY (`id`),
  9.   UNIQUE KEY `order_id_index` (`order_id`) USING BTREE
  10. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

SPI配置

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/757119
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号