使用ShardingSphere实现分库分表及历史数据处理方案-ZHD的小窝

ShardingSphere是Apache顶级开源项目，提供了一套完整的分库分表解决方案。下面详细介绍如何使用ShardingSphere实现分库分表并处理历史数据。

一、ShardingSphere分库分表基础配置

1. 添加依赖

<!-- Spring Boot Starter -->
<dependency>
    <groupId>org.apache.shardingsphere</groupId>
    <artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId>
    <version>5.3.2</version>
</dependency>

2. 基础配置示例（application.yml）

spring:
  shardingsphere:
    datasource:
      names: ds0,ds1
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/db0
        username: root
        password: root
      ds1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/db1
        username: root
        password: root
    rules:
      sharding:
        tables:
          t_order:
            actual-data-nodes: ds$->{0..1}.t_order_$->{0..15}
            table-strategy:
              standard:
                sharding-column: order_id
                precise-algorithm-class-name: com.example.MyPreciseShardingAlgorithm
            key-generate-strategy:
              column: order_id
              key-generator-name: snowflake
        key-generators:
          snowflake:
            type: SNOWFLAKE
            props:
              worker-id: 123

3. 自定义分片算法

public class MyPreciseShardingAlgorithm implements PreciseShardingAlgorithm<Long> {
    @Override
    public String doSharding(Collection<String> availableTargetNames, PreciseShardingValue<Long> shardingValue) {
        // 根据order_id计算分表后缀
        long orderId = shardingValue.getValue();
        String tableSuffix = String.valueOf(orderId % 16);
        for (String each : availableTargetNames) {
            if (each.endsWith(tableSuffix)) {
                return each;
            }
        }
        throw new UnsupportedOperationException();
    }
}

二、历史数据处理方案

1. 使用ShardingSphere-Scaling进行数据迁移

ShardingSphere提供了Scaling模块专门用于数据迁移：

# scaling配置示例
scaling:
  name: defaultScaling
  workerThread: 40
  batchSize: 1000
  rateLimiter:
    type: QPS
    props:
      qps: 50

迁移流程：

创建迁移任务
执行全量数据迁移
执行增量数据同步
切换业务流量

2. 迁移代码示例

@RestController
@RequestMapping("/scaling")
public class ScalingController {
    
    @Autowired
    private ScalingAPI scalingAPI;
    
    @PostMapping("/start")
    public String startMigration() {
        JobConfiguration jobConfig = new JobConfiguration();
        jobConfig.setJobId("migration-job-1");
        jobConfig.setSourceResourceName("ds_old");
        jobConfig.setSourceSchemaName("db_old");
        jobConfig.setTargetResourceName("ds_new");
        jobConfig.setTargetSchemaName("db_new");
        
        // 配置表映射规则
        jobConfig.getJobShardingDataNodes().add(
            new JobShardingDataNode("t_order", "ds0.t_order_0,ds0.t_order_1,...")
        );
        
        scalingAPI.start(jobConfig);
        return "Migration started";
    }
    
    @GetMapping("/progress/{jobId}")
    public JobProgress getProgress(@PathVariable String jobId) {
        return scalingAPI.getProgress(jobId);
    }
}

3. 双写过渡方案

在迁移期间，可以使用ShardingSphere的HintManager实现强制路由，确保新旧数据一致：

@Service
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Transactional
    public void createOrder(Order order) {
        // 写入新分片表
        orderRepository.save(order);
        
        // 使用HintManager强制写入旧表
        try (HintManager hintManager = HintManager.getInstance()) {
            hintManager.setDatabaseShardingValue("ds_old");
            orderRepository.save(order);
        }
    }
}

4. 历史数据归档策略

# 配置归档规则
spring:
  shardingsphere:
    rules:
      sharding:
        tables:
          t_order:
            actual-data-nodes: ds$->{0..1}.t_order_$->{0..15}
            auto-table:
              t_order_archive:
                actual-data-sources: ds_archive
                sharding-strategy:
                  none:

三、查询历史数据方案

1. 使用ShardingSphere的绑定表

spring:
  shardingsphere:
    rules:
      sharding:
        binding-tables:
          - t_order,t_order_archive

2. 联邦查询配置

-- 创建逻辑视图
CREATE SHARDING VIEW order_view AS 
SELECT * FROM t_order 
UNION ALL 
SELECT * FROM t_order_archive;

3. 多数据源查询

@Repository
public class OrderDao {
    
    @Autowired
    @Qualifier("newDataSource")
    private DataSource newDataSource;
    
    @Autowired
    @Qualifier("oldDataSource")
    private DataSource oldDataSource;
    
    public List<Order> findHistoricalOrders(Long userId) {
        // 查询新分片表
        List<Order> newOrders = queryFromDataSource(newDataSource, userId);
        
        // 查询旧单表
        List<Order> oldOrders = queryFromDataSource(oldDataSource, userId);
        
        // 合并结果
        List<Order> allOrders = new ArrayList<>();
        allOrders.addAll(newOrders);
        allOrders.addAll(oldOrders);
        return allOrders;
    }
}

四、最佳实践建议

迁移前准备：
- 评估数据量和业务影响
- 准备完善的回滚方案
- 在测试环境充分验证
迁移期间：
- 选择业务低峰期执行
- 监控迁移进度和系统性能
- 保持新旧系统双写一段时间
迁移后：
- 验证数据一致性
- 性能基准测试
- 逐步淘汰旧系统
长期维护：
- 建立定期归档机制
- 监控分片均衡性
- 准备扩容方案

通过ShardingSphere的这些功能，您可以相对平滑地实现分库分表改造，并妥善处理历史数据问题。

目录CONTENT

使用ShardingSphere实现分库分表及历史数据处理方案