数据分区与复制

数据分区（Sharding）和数据复制（Replication）是分布式存储系统的两大核心技术。分区将数据分散到多个节点，突破单节点的存储和计算瓶颈；复制将数据复制到多个副本，提高数据的可用性和读取性能。两者结合，构建出可扩展、高可用的分布式存储系统。

为什么需要数据分区

单机数据库在面对大规模数据和高并发访问时，会遇到明显的瓶颈。

单机的物理限制

存储容量限制：单台服务器的硬盘容量有限。即使使用多块硬盘组建 RAID，容量扩展也有上限。当数据量达到 TB 甚至 PB 级别时，单机存储不再可行。

计算能力限制：单个 CPU 的核心数有限，处理能力有限。当并发查询数量激增时，CPU 成为瓶颈，查询响应时间急剧增加。

内存容量限制：数据库的性能很大程度上依赖于缓存。单机内存有限，无法缓存全部热点数据，导致频繁的磁盘 I/O，性能下降。

网络带宽限制：单机的网络接口带宽有限，当读写请求量超过带宽上限时，延迟会显著增加。

扩展的经济性：纵向扩展（Scale Up，升级硬件）的成本呈指数增长。一台性能翻倍的服务器，价格可能翻几倍。而横向扩展（Scale Out，增加服务器）的成本增长相对线性。

数据分区策略

数据分区的核心问题是：如何将数据均匀地分散到多个节点，同时支持高效的查询？

哈希分区

哈希分区是最常用的分区策略。对数据的某个字段（通常是主键）计算哈希值，然后对分区数取模，决定数据所属的分区。

// 哈希分区实现
public class HashSharding {
    
    private final int shardCount;
    
    /**
     * 计算数据所属的分片
     * @param key 数据键（如用户ID）
     * @return 分片编号
     */
    public int getShard(String key) {
        // 使用 MurmurHash 比 Java hashCode 更均匀
        int hash = MurmurHash.hash32(key);
        // 取模确定分片
        return Math.abs(hash) % shardCount;
    }
}

优点：

数据分布均匀（假设哈希函数足够好）
点查询效率高，一次计算即可定位分区
实现简单，易于理解

缺点：

不支持范围查询（相邻的键可能分布在不同的分区）
增减分区时需要大量数据迁移（所有数据都需要重新计算哈希）

范围分区

范围分区按照数据的键值范围划分分区。每个分区负责一段连续的键范围。

// 范围分区实现
public class RangeSharding {
    
    private final List<RangeShard> shards;
    
    /**
     * 范围分区映射表
     */
    static class RangeShard {
        String start;  // 范围起点（包含）
        String end;    // 范围终点（不包含）
        int shardId;   // 分片编号
    }
    
    /**
     * 计算数据所属的分片
     */
    public int getShard(String key) {
        // 二分查找确定范围
        for (RangeShard shard : shards) {
            if (key.compareTo(shard.start) >= 0 && 
                key.compareTo(shard.end) < 0) {
                return shard.shardId;
            }
        }
        // 默认返回最后一个分区
        return shards.get(shards.size() - 1).shardId;
    }
}

优点：

支持范围查询（相邻的键在同一个分区）
增减分区相对简单（只需调整范围边界）
适合按时间顺序存储的数据（如日志、时间序列）

缺点：

可能出现数据倾斜（某些范围的数据量远大于其他范围）
热点问题（某些范围被频繁访问，如最新的数据）

3. 一致性哈希

// 一致性哈希实现
class ConsistentHash {
    
    // 虚拟节点数量
    private static final int VIRTUAL_NODES = 150;
    
    // 哈希环
    private final SortedMap<Integer, String> circle = new TreeMap<>();
    
    public ConsistentHash(List<String> nodes) {
        for (String node : nodes) {
            addNode(node);
        }
    }
    
    // 添加节点
    public void addNode(String node) {
        for (int i = 0; i < VIRTUAL_NODES; i++) {
            String vNode = node + "#" + i;
            int hash = hash(vNode);
            circle.put(hash, node);
        }
    }
    
    // 移除节点
    public void removeNode(String node) {
        for (int i = 0; i < VIRTUAL_NODES; i++) {
            String vNode = node + "#" + i;
            int hash = hash(vNode);
            circle.remove(hash);
        }
    }
    
    // 获取数据对应的节点
    public String getNode(String key) {
        if (circle.isEmpty()) {
            throw new IllegalStateException("没有可用节点");
        }
        
        int hash = hash(key);
        
        // 找到第一个大于等于hash的节点
        SortedMap<Integer, String> tail = circle.tailMap(hash);
        
        if (tail.isEmpty()) {
            // 环回到开头
            return circle.get(circle.firstKey());
        }
        
        return tail.get(tail.firstKey());
    }
    
    private int hash(String key) {
        return Math.abs(key.hashCode());
    }
}

分区架构

分区带来的问题

跨分区查询

// 跨分片查询处理
class CrossShardQuery {
    
    // 聚合查询：需要从多个分片获取结果后聚合
    public int countUsersByAge(int age) {
        int totalCount = 0;
        
        // 并行查询所有分片
        List<Future<Integer>> futures = new ArrayList<>();
        for (Shard shard : allShards) {
            futures.add(executor.submit(() -> shard.countByAge(age)));
        }
        
        // 聚合结果
        for (Future<Integer> future : futures) {
            totalCount += future.get();
        }
        
        return totalCount;
    }
    
    // 排序查询：从多分片获取数据后排序
    public List<User> getTopUsers(int limit) {
        List<User> allUsers = new ArrayList<>();
        
        // 并行获取各分片数据
        for (Shard shard : allShards) {
            allUsers.addAll(shard.getAllUsers());
        }
        
        // 内存排序取Top N
        return allUsers.stream()
            .sorted(Comparator.comparing(User::getScore).reversed())
            .limit(limit)
            .collect(Collectors.toList());
    }
}

分布式ID

// 分布式ID生成方案
class DistributedIdGenerator {
    
    // 方案1：UUID
    public String generateUUID() {
        return UUID.randomUUID().toString();
    }
    
    // 方案2：雪花算法
    static class SnowflakeId {
        // 64位: 1位(符号) + 41位(时间戳) + 10位(机器ID) + 12位(序列号)
        private long sequence = 0;
        private long lastTimestamp = -1;
        private final long workerId;
        
        public SnowflakeId(long workerId) {
            this.workerId = workerId;
        }
        
        public synchronized long nextId() {
            long timestamp = System.currentTimeMillis();
            
            // 时钟回拨处理
            if (timestamp < lastTimestamp) {
                timestamp = lastTimestamp;
            }
            
            if (timestamp == lastTimestamp) {
                // 同一毫秒内，序列号+1
                sequence = (sequence + 1) & ((1 << 12) - 1);
                if (sequence == 0) {
                    // 序列号用完，等待下一毫秒
                    timestamp = waitNextMillis(timestamp);
                }
            } else {
                sequence = 0;
            }
            
            lastTimestamp = timestamp;
            
            // 组装ID
            return ((timestamp - 1609459200000L) << 22) |
                   (workerId << 12) |
                   sequence;
        }
    }
    
    // 方案3：数据库号段
    static class DatabaseSegmentId {
        private static final long BATCH_SIZE = 1000;
        private long currentId = 0;
        private long maxId = 0;
        
        public synchronized long nextId() {
            if (currentId >= maxId) {
                // 从数据库获取新的号段
                currentId = fetchFromDatabase();
                maxId = currentId + BATCH_SIZE;
            }
            return currentId++;
        }
    }
}

数据复制

复制模式

1. 主从复制

// 主从复制实现
class MasterSlaveReplication {
    
    // 主节点写入
    public void write(String key, String value) {
        // 1. 写入主节点
        masterDB.write(key, value);
        
        // 2. 同步/异步复制到从节点
        replicateToSlaves(key, value);
    }
    
    // 从节点读取
    public String read(String key, ReadConsistency consistency) {
        if (consistency == ReadConsistency.STRONG) {
            // 强一致性：读取主节点或多数派
            return readFromQuorum(key);
        } else {
            // 弱一致性：读取任意从节点
            return readFromAnySlave(key);
        }
    }
    
    // 异步复制
    private void replicateToSlaves(String key, String value) {
        for (SlaveNode slave : slaves) {
            // 异步复制，不等待完成
            slave.asyncReplicate(key, value);
        }
    }
}

2. 复制策略

策略	优点	缺点	适用场景
同步复制	数据不丢失	延迟高	金融交易
异步复制	延迟低	可能丢数据	普通业务
半同步复制	平衡	-	大多数场景

// 半同步复制
class SemiSyncReplication {
    
    public void write(String key, String value) {
        // 1. 写入主节点
        masterDB.write(key, value);
        
        // 2. 同步复制到至少N个节点
        int syncCount = 0;
        for (ReplicaNode replica : replicas) {
            if (replica.syncWrite(key, value)) {
                syncCount++;
            }
            // 多数派确认后返回
            if (syncCount >= requiredQuorum) {
                break;
            }
        }
        
        // 3. 剩余节点异步复制
        for (ReplicaNode replica : replicas) {
            if (!replica.isSynced()) {
                replica.asyncReplicate(key, value);
            }
        }
    }
}

3. 复制延迟处理

// 复制延迟处理
class ReplicationLagHandler {
    
    // 读请求重定向
    public String readWithRetry(String key) {
        try {
            // 优先读从节点
            return readFromSlave(key);
        } catch (StaleDataException e) {
            // 数据过期，重试读主节点
            return readFromMaster(key);
        }
    }
    
    // 写入后立即读取
    public void writeAndRead(String key, String value) {
        // 1. 写入主节点
        masterDB.write(key, value);
        
        // 2. 同步复制到足够节点
        replicateToQuorum(key, value);
        
        // 3. 确保复制完成后再读取
        waitForReplication(key);
        
        // 4. 读取
        return readFromMaster(key);
    }
}

数据一致性模型

// 各种一致性模型实现

// 1. 强一致性：所有节点同步
class StrongConsistency {
    public void write(String key, String value) {
        // 同步到所有节点
        for (Node node : allNodes) {
            node.syncWrite(key, value);
        }
    }
}

// 2. 顺序一致性：保持操作顺序
class SequentialConsistency {
    // 使用向量时钟保证顺序
    VectorClock clock = new VectorClock();
    
    public void write(String key, String value) {
        // 为操作分配全局顺序
        long sequence = clock.incrementAndGet();
        node.write(key, value, sequence);
    }
}

// 3. 因果一致性：保证因果关系
class CausalConsistency {
    // 记录因果关系
    Map<String, VectorClock> causality = new HashMap<>();
    
    public void write(String key, String value, VectorClock dependsOn) {
        // 检查依赖是否满足
        if (dependsOn != null && !isCausalitySatisfied(dependsOn)) {
            // 等待依赖操作完成
            waitFor(dependsOn);
        }
        node.write(key, value);
    }
}

// 4. 最终一致性：允许短暂不一致
class EventualConsistency {
    public void write(String key, String value) {
        // 立即返回，异步复制
        node.writeLocal(key, value);
        asyncReplicateToAll(key, value);
    }
}