05. 一致性哈希与数据分片

import hashlib
import bisect

class ConsistentHashRing:
    def __init__(self, replicas=150):
        self.replicas = replicas  # 每个物理节点的虚拟节点数
        self.ring = {}            # {hash: node}
        self.sorted_keys = []     # 排序的哈希值
        self.nodes = set()
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode('utf-8')).hexdigest(), 16)
    
    def add_node(self, node):
        self.nodes.add(node)
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            bisect.insort(self.sorted_keys, key)
    
    def remove_node(self, node):
        self.nodes.discard(node)
        for i in range(self.replicas):
            key = self._hash(f"{node}:{i}")
            del self.ring[key]
            idx = bisect.bisect_left(self.sorted_keys, key)
            self.sorted_keys.pop(idx)
    
    def get_node(self, key):
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]
    
    def get_nodes(self, key, n=3):
        """获取顺时针的 n 个节点（用于多副本）"""
        if not self.ring or n > len(self.nodes):
            return list(self.nodes)
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h)
        result = []
        seen = set()
        while len(result) < n:
            node = self.ring[self.sorted_keys[idx % len(self.sorted_keys)]]
            if node not in seen:
                seen.add(node)
                result.append(node)
            idx += 1
        return result

# 使用
ch = ConsistentHashRing(replicas=150)
ch.add_node("192.168.1.1")
ch.add_node("192.168.1.2")
ch.add_node("192.168.1.3")

node = ch.get_node("user:12345")
replicas = ch.get_nodes("order:67890", n=3)  # 3 个副本节点

# 元数据表
shard_map = {
    'user:1': 'shard_0',
    'user:2': 'shard_1',
    # ...
}

# 查找
shard = shard_map.get('user:123') or compute_shard('user:123')

def migrate_data(consistent_hash, new_node):
    """迁移应落在 new_node 上的数据"""
    # 1. 将 new_node 加入环（双写阶段）
    consistent_hash.add_node(new_node)
    
    # 2. 遍历数据，将归属变为 new_node 的数据复制过去
    for key, value in source_db.scan():
        assigned = consistent_hash.get_node(key)
        if assigned == new_node:
            new_node.store(key, value)
    
    # 3. 验证一致性后，确认新节点服务读请求
    # 4. 删除旧节点上的冗余数据

# 查看集群槽分配
redis-cli -c -p 7000 cluster slots

# 迁移槽
redis-cli --cluster reshard 127.0.0.1:7000 \
  --cluster-from node-id-a \
  --cluster-to node-id-b \
  --cluster-slots 1000

-- 创建 keyspace 指定复制策略
CREATE KEYSPACE mykeyspace
WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'datacenter1': 3
};