数据库分片实战：水平扩展的架构设计与实现

// sharding/HashShardingStrategy.java
public class HashShardingStrategy implements ShardingStrategy {
    private final int shardCount;
    
    public HashShardingStrategy(int shardCount) {
        if (shardCount <= 0 || (shardCount & (shardCount - 1)) != 0) {
            throw new IllegalArgumentException("Shard count must be a power of 2");
        }
        this.shardCount = shardCount;
    }
    
    @Override
    public int calculateShard(Object shardKey) {
        int hash = MurmurHash3.hash32(shardKey.toString());
        // 使用位运算代替取模，性能更好
        return hash & (shardCount - 1);
    }
    
    @Override
    public List<Integer> calculateShards(Range<Object> range) {
        // 范围查询需要访问所有分片
        return IntStream.range(0, shardCount).boxed().collect(Collectors.toList());
    }
}

// 使用示例
HashShardingStrategy strategy = new HashShardingStrategy(16);
int shardId = strategy.calculateShard(userId); // 0-15
String tableName = "users_" + shardId; // users_0, users_1, ...

// sharding/RangeShardingStrategy.java
public class RangeShardingStrategy implements ShardingStrategy {
    private final TreeMap<Long, Integer> rangeMap;
    
    public RangeShardingStrategy() {
        this.rangeMap = new TreeMap<>();
        // 按时间范围分片
        rangeMap.put(0L, 0);                    // 2020年之前 -> shard 0
        rangeMap.put(1577836800000L, 1);        // 2020年 -> shard 1
        rangeMap.put(1609459200000L, 2);        // 2021年 -> shard 2
        rangeMap.put(1640995200000L, 3);        // 2022年 -> shard 3
    }
    
    @Override
    public int calculateShard(Object shardKey) {
        long timestamp = (Long) shardKey;
        return rangeMap.floorEntry(timestamp).getValue();
    }
    
    @Override
    public List<Integer> calculateShards(Range<Object> range) {
        long start = (Long) range.lowerEndpoint();
        long end = (Long) range.upperEndpoint();
        
        Set<Integer> shards = new HashSet<>();
        for (Map.Entry<Long, Integer> entry : rangeMap.subMap(start, true, end, true).entrySet()) {
            shards.add(entry.getValue());
        }
        
        return new ArrayList<>(shards);
    }
}

// 使用示例：订单表按年月分片
CREATE TABLE orders_202401 (
    id BIGINT PRIMARY KEY,
    user_id BIGINT,
    order_date DATE,
    amount DECIMAL(10,2)
);

CREATE TABLE orders_202402 (...);

// sharding/DirectoryShardingStrategy.java
public class DirectoryShardingStrategy implements ShardingStrategy {
    private final Map<Object, Integer> directory;
    private final DataSource configDb;
    
    public DirectoryShardingStrategy(DataSource configDb) {
        this.configDb = configDb;
        this.directory = loadDirectory();
    }
    
    private Map<Object, Integer> loadDirectory() {
        Map<Object, Integer> dir = new HashMap<>();
        
        try (Connection conn = configDb.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(
                 "SELECT shard_key, shard_id FROM shard_directory")) {
            
            while (rs.next()) {
                dir.put(rs.getString("shard_key"), rs.getInt("shard_id"));
            }
        }
        
        return dir;
    }
    
    @Override
    public int calculateShard(Object shardKey) {
        Integer shardId = directory.get(shardKey);
        if (shardId == null) {
            throw new ShardingException("Shard key not found: " + shardKey);
        }
        return shardId;
    }
    
    public void updateDirectory(Object shardKey, int newShardId) {
        directory.put(shardKey, newShardId);
        
        // 持久化到配置库
        try (Connection conn = configDb.getConnection();
             PreparedStatement stmt = conn.prepareStatement(
                 "INSERT INTO shard_directory (shard_key, shard_id) VALUES (?, ?) " +
                 "ON DUPLICATE KEY UPDATE shard_id = ?")) {
            
            stmt.setString(1, shardKey.toString());
            stmt.setInt(2, newShardId);
            stmt.setInt(3, newShardId);
            stmt.executeUpdate();
        }
    }
}

# sharding-config.yaml
spring:
  shardingsphere:
    datasource:
      names: ds0,ds1,ds2,ds3
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/order_db_0
        username: root
        password: password
      ds1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/order_db_1
        username: root
        password: password
      # ds2, ds3 类似配置
    
    rules:
      sharding:
        tables:
          # 订单表分片
          t_order:
            actual-data-nodes: ds$->{0..3}.t_order_$->{0..7}
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: order-db-hash
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: order-table-hash
            key-generate-strategy:
              column: order_id
              key-generator-name: snowflake
          
          # 订单详情表（与订单表绑定）
          t_order_item:
            actual-data-nodes: ds$->{0..3}.t_order_item_$->{0..7}
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: order-db-hash
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: order-table-hash
        
        sharding-algorithms:
          order-db-hash:
            type: HASH_MOD
            props:
              sharding-count: 4
          order-table-hash:
            type: HASH_MOD
            props:
              sharding-count: 8
        
        key-generators:
          snowflake:
            type: SNOWFLAKE
            props:
              worker-id: 123

// service/OrderQueryService.java
@Service
public class OrderQueryService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    // 单分片查询（指定分片键）
    public Order getOrder(Long orderId, Long userId) {
        // ShardingSphere 自动路由到正确的分片
        return orderRepository.findByOrderIdAndUserId(orderId, userId);
    }
    
    // 跨分片查询（未指定分片键）
    public List<Order> getOrdersByStatus(String status) {
        // 需要扫描所有分片，性能较差
        return orderRepository.findByStatus(status);
    }
    
    // 优化：使用二级索引表
    @Autowired
    private OrderIndexRepository orderIndexRepository;
    
    public List<Order> getOrdersByStatusOptimized(String status, int limit) {
        // 1. 从索引表获取订单ID和分片键
        List<OrderIndex> indexes = orderIndexRepository
            .findByStatus(status, limit);
        
        // 2. 按分片分组
        Map<Integer, List<Long>> ordersByShard = indexes.stream()
            .collect(Collectors.groupingBy(
                idx -> calculateShard(idx.getUserId()),
                Collectors.mapping(OrderIndex::getOrderId, Collectors.toList())
            ));
        
        // 3. 并行查询各分片
        return ordersByShard.entrySet().parallelStream()
            .flatMap(entry -> {
                List<Long> orderIds = entry.getValue();
                return orderRepository.findByOrderIds(orderIds).stream();
            })
            .collect(Collectors.toList());
    }
}

// migration/ShardMigrationService.java
@Service
public class ShardMigrationService {
    
    @Autowired
    private DataSource sourceDs;
    
    @Autowired
    private DataSource targetDs;
    
    public void migrateShard(int sourceShard, int targetShard, 
                            Predicate<Long> shouldMigrate) {
        int batchSize = 1000;
        long lastId = 0;
        
        while (true) {
            // 1. 从源分片读取一批数据
            List<Order> batch = readBatch(sourceShard, lastId, batchSize);
            if (batch.isEmpty()) {
                break;
            }
            
            // 2. 过滤需要迁移的数据
            List<Order> toMigrate = batch.stream()
                .filter(order -> shouldMigrate.test(order.getUserId()))
                .collect(Collectors.toList());
            
            // 3. 写入目标分片
            if (!toMigrate.isEmpty()) {
                writeBatch(targetShard, toMigrate);
                
                // 4. 从源分片删除
                deleteBatch(sourceShard, toMigrate);
            }
            
            lastId = batch.get(batch.size() - 1).getId();
            
            // 控制迁移速度
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    private List<Order> readBatch(int shard, long lastId, int limit) {
        String sql = "SELECT * FROM t_order_" + shard + 
                     " WHERE id > ? ORDER BY id LIMIT ?";
        
        try (Connection conn = sourceDs.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            
            stmt.setLong(1, lastId);
            stmt.setInt(2, limit);
            
            List<Order> orders = new ArrayList<>();
            try (ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) {
                    orders.add(mapOrder(rs));
                }
            }
            return orders;
        }
    }
}