上一讲我们实现了节点发现和网络通信，三个节点能互相感知了。但有个核心问题没解决：数据该存到哪个节点上？

最简单的做法是取模：hash(key) % N。但如果节点数变化（扩容或宕机），几乎所有数据的映射关系都会改变，导致大规模数据迁移。

这一讲，我们实现一致性哈希——一种让数据迁移量最小化的分片策略。

一、一致性哈希原理

1.1 核心思想

传统取模：key → hash(key) % N ❌ 节点增减时，几乎所有 key 都需要重新映射 一致性哈希：key → hash(key) → 环上的第一个节点 ✅ 节点增减时，只有少量 key 需要迁移

哈希环：

hash(node1) ▲ │ hash(k4) │ hash(k1) ◄─────┼──────► │ hash(k3) │ hash(k2) │ ▼ hash(node2)

每个 key 顺时针找到的第一个节点，就是它所属的节点。

1.2 虚拟节点

为了解决两个问题：

1. 数据倾斜：节点在环上分布不均匀，导致某些节点数据过多

2. 负载不均：不同节点性能不同，需要差异化权重

引入虚拟节点：每个物理节点对应多个虚拟节点，分散在环上。

物理节点 A → 虚拟节点 A-0, A-1, A-2, ..., A-159 物理节点 B → 虚拟节点 B-0, B-1, B-2, ..., B-159 物理节点 C → 虚拟节点 C-0, C-1, C-2, ..., C-159

二、一致性哈希实现

2.1 哈希环

# minikv/hashring/consistent_hash.py import hashlib import bisect from typing import Dict, List, Optional, Callable from dataclasses import dataclass, field class HashRing: """ 一致性哈希环 支持： - 虚拟节点（解决数据倾斜） - 权重（不同节点不同容量） - 节点增删（最小化数据迁移） """ def __init__(self, nodes: List[str] = None, virtual_nodes: int = 160, hash_fn: Callable = None): """ Args: nodes: 初始节点列表 virtual_nodes: 每个物理节点的虚拟节点数 hash_fn: 哈希函数，默认使用 MD5 """ self.virtual_nodes = virtual_nodes self.hash_fn = hash_fn or self._md5_hash # 环：有序的 (hash_value, virtual_node_id) 列表 self.ring: List[tuple] = [] # 虚拟节点 → 物理节点 映射 self.virtual_to_physical: Dict[str, str] = {} # 物理节点 → 虚拟节点列表 映射 self.physical_to_virtuals: Dict[str, List[str]] = {} # 节点权重 self.weights: Dict[str, int] = {} if nodes: for node in nodes: self.add_node(node) def _md5_hash(self, key: str) -> int: """MD5 哈希，返回 128 位整数""" return int(hashlib.md5(key.encode('utf-8')).hexdigest(), 16) def _build_virtual_key(self, node: str, vnode_index: int) -> str: """生成虚拟节点的 key""" return f"{node}#{vnode_index}" def add_node(self, node: str, weight: int = 1): """ 添加物理节点 Args: node: 节点标识符（如 'node-1'） weight: 权重，越大表示该节点承担的负载越多 """ if node in self.physical_to_virtuals: return # 节点已存在 self.weights[node] = weight self.physical_to_virtuals[node] = [] # 根据权重创建虚拟节点 actual_vnodes = self.virtual_nodes * weight for i in range(actual_vnodes): vnode_key = self._build_virtual_key(node, i) hash_val = self.hash_fn(vnode_key) # 插入到有序环中 bisect.insort(self.ring, (hash_val, vnode_key)) self.virtual_to_physical[vnode_key] = node self.physical_to_virtuals[node].append(vnode_key) def remove_node(self, node: str): """ 移除物理节点及其所有虚拟节点 """ if node not in self.physical_to_virtuals: return for vnode_key in self.physical_to_virtuals[node]: hash_val = self.hash_fn(vnode_key) # 从环中移除 idx = bisect.bisect_left(self.ring, (hash_val, vnode_key)) if idx < len(self.ring) and self.ring[idx] == (hash_val, vnode_key): self.ring.pop(idx) del self.virtual_to_physical[vnode_key] del self.physical_to_virtuals[node] del self.weights[node] def get_node(self, key: str) -> Optional[str]: """ 获取 key 所属的物理节点 在环上顺时针查找第一个虚拟节点 """ if not self.ring: return None hash_val = self.hash_fn(key) # 二分查找第一个大于等于 hash_val 的位置 idx = bisect.bisect_left(self.ring, (hash_val, '')) # 如果超出环尾，回到环首 if idx >= len(self.ring): idx = 0 vnode_key = self.ring[idx][1] return self.virtual_to_physical.get(vnode_key) def get_nodes(self, key: str, count: int = 1) -> List[str]: """ 获取 key 所属的多个物理节点（用于副本放置） 返回顺时针方向上的 count 个不同物理节点 """ if not self.ring or count <= 0: return [] result = [] seen = set() hash_val = self.hash_fn(key) idx = bisect.bisect_left(self.ring, (hash_val, '')) # 从 idx 开始顺时针遍历 for i in range(len(self.ring)): ring_idx = (idx + i) % len(self.ring) vnode_key = self.ring[ring_idx][1] physical_node = self.virtual_to_physical[vnode_key] if physical_node not in seen: seen.add(physical_node) result.append(physical_node) if len(result) >= count: break return result def get_node_load(self) -> Dict[str, float]: """ 获取每个物理节点的理论负载比例 """ if not self.ring: return {} total_vnodes = len(self.ring) load = {} for vnode_key in self.virtual_to_physical: physical = self.virtual_to_physical[vnode_key] load[physical] = load.get(physical, 0) + 1 return {k: v / total_vnodes for k, v in load.items()} def get_nodes_count(self) -> int: """获取物理节点数""" return len(self.physical_to_virtuals) def __str__(self): nodes = list(self.physical_to_virtuals.keys()) return f"HashRing(nodes={nodes}, total_vnodes={len(self.ring)})"

2.2 带副本的一致性哈希

# minikv/hashring/replicated_hash.py from typing import List, Optional from .consistent_hash import HashRing class ReplicatedHashRing(HashRing): """ 带副本的一致性哈希环 每个 key 映射到多个节点，实现数据冗余 """ def __init__(self, nodes: List[str] = None, replication_factor: int = 3, virtual_nodes: int = 160): """ Args: nodes: 初始节点列表 replication_factor: 副本数 virtual_nodes: 每个物理节点的虚拟节点数 """ super().__init__(nodes, virtual_nodes) self.replication_factor = replication_factor def get_replica_nodes(self, key: str) -> List[str]: """ 获取 key 的所有副本节点 返回 replication_factor 个不同的物理节点 """ return self.get_nodes(key, self.replication_factor) def is_primary(self, key: str, node: str) -> bool: """ 判断指定节点是否是 key 的主节点 """ primary = self.get_node(key) return primary == node def is_replica(self, key: str, node: str) -> bool: """ 判断指定节点是否是 key 的副本节点 """ replicas = self.get_replica_nodes(key) return node in replicas

三、数据分布管理器

3.1 分片管理器

# minikv/hashring/shard_manager.py import threading import logging from typing import Dict, List, Optional, Set, Callable from dataclasses import dataclass, field from .replicated_hash import ReplicatedHashRing logger = logging.getLogger(__name__) @dataclass class ShardInfo: """分片信息""" shard_id: str primary_node: str replica_nodes: List[str] key_range_start: str = "" key_range_end: str = "" size_bytes: int = 0 key_count: int = 0 class ShardManager: """ 分片管理器 负责： 1. 数据分布计算 2. 分片迁移决策 3. 负载均衡 """ def __init__(self, local_node_id: str, replication_factor: int = 3, virtual_nodes: int = 160): self.local_node_id = local_node_id self.replication_factor = replication_factor self.hash_ring = ReplicatedHashRing( replication_factor=replication_factor, virtual_nodes=virtual_nodes ) # 分片信息 self.shards: Dict[str, ShardInfo] = {} # 本节点负责的分片 self.local_shards: Set[str] = set() # 迁移中的分片 self.migrating_shards: Dict[str, str] = {} # shard_id -> target_node self.lock = threading.RLock() # 回调 self.on_shard_moved: Optional[Callable] = None def add_node(self, node_id: str, weight: int = 1): """添加节点""" with self.lock: self.hash_ring.add_node(node_id, weight) self._rebalance_local_shards() logger.info(f"添加节点: {node_id}") def remove_node(self, node_id: str): """移除节点""" with self.lock: # 记录需要迁移的分片 shards_to_move = [ sid for sid, info in self.shards.items() if info.primary_node == node_id ] self.hash_ring.remove_node(node_id) # 触发迁移 for shard_id in shards_to_move: self._migrate_shard(shard_id) logger.info(f"移除节点: {node_id}") def get_shard_for_key(self, key: str) -> Optional[ShardInfo]: """ 获取 key 所在的分片 """ primary = self.hash_ring.get_node(key) if not primary: return None # 查找对应的分片 for shard in self.shards.values(): if shard.primary_node == primary: return shard return None def get_replicas_for_key(self, key: str) -> List[str]: """ 获取 key 的副本节点列表 """ return self.hash_ring.get_replica_nodes(key) def should_handle_key(self, key: str) -> bool: """ 判断本节点是否应该处理该 key """ return self.hash_ring.is_replica(key, self.local_node_id) def _rebalance_local_shards(self): """重新平衡本节点负责的分片""" # 获取本节点现在应该负责的所有 key 范围 # 简化实现：重新计算所有分片归属 old_shards = self.local_shards.copy() new_shards = set() for shard_id, info in self.shards.items(): new_primary = self.hash_ring.get_node(shard_id) if new_primary == self.local_node_id: new_shards.add(shard_id) self.local_shards = new_shards # 记录需要迁出的分片 shards_to_remove = old_shards - new_shards for shard_id in shards_to_remove: logger.info(f"分片 {shard_id} 将从本节点迁出") def _migrate_shard(self, shard_id: str): """迁移分片到新的主节点""" shard = self.shards.get(shard_id) if not shard: return new_primary = self.hash_ring.get_node(shard_id) if new_primary == shard.primary_node: return # 不需要迁移 self.migrating_shards[shard_id] = new_primary logger.info(f"开始迁移分片 {shard_id}: {shard.primary_node} → {new_primary}") # 更新分片信息 shard.primary_node = new_primary if self.on_shard_moved: self.on_shard_moved(shard_id, new_primary) del self.migrating_shards[shard_id] def get_distribution(self) -> Dict[str, Dict]: """ 获取数据分布情况 """ distribution = {} for node_id in self.hash_ring.physical_to_virtuals: shards_for_node = [ sid for sid, info in self.shards.items() if info.primary_node == node_id ] distribution[node_id] = { 'shard_count': len(shards_for_node), 'shards': shards_for_node, 'load': self.hash_ring.get_node_load().get(node_id, 0) } return distribution def get_stats(self) -> dict: """获取统计信息""" with self.lock: return { 'nodes': self.hash_ring.get_nodes_count(), 'shards': len(self.shards), 'local_shards': len(self.local_shards), 'migrating': len(self.migrating_shards), 'distribution': self.get_distribution() }

四、负载均衡器

4.1 自适应负载均衡

# minikv/hashring/load_balancer.py import threading import time import logging from typing import Dict, List, Optional from dataclasses import dataclass from .shard_manager import ShardManager logger = logging.getLogger(__name__) @dataclass class NodeLoad: """节点负载信息""" node_id: str cpu_usage: float = 0.0 memory_usage: float = 0.0 disk_usage: float = 0.0 request_count: int = 0 avg_latency_ms: float = 0.0 shard_count: int = 0 class LoadBalancer: """ 自适应负载均衡器 根据节点负载动态调整权重和分片分布 """ def __init__(self, shard_manager: ShardManager, balance_interval: float = 30.0, max_load_diff: float = 0.2): self.shard_manager = shard_manager self.balance_interval = balance_interval self.max_load_diff = max_load_diff self.node_loads: Dict[str, NodeLoad] = {} self.running = False self.lock = threading.Lock() def start(self): """启动负载均衡""" self.running = True thread = threading.Thread(target=self._balance_loop, daemon=True) thread.start() logger.info("负载均衡器启动") def stop(self): """停止负载均衡""" self.running = False def report_load(self, node_id: str, load: NodeLoad): """上报节点负载""" with self.lock: self.node_loads[node_id] = load def _balance_loop(self): """负载均衡循环""" while self.running: time.sleep(self.balance_interval) self._try_balance() def _try_balance(self): """尝试进行负载均衡""" with self.lock: if len(self.node_loads) < 2: return # 找出最忙和最闲的节点 loads = list(self.node_loads.values()) loads.sort(key=lambda x: x.request_count) min_load = loads[0] max_load = loads[-1] # 如果负载差异超过阈值，触发 rebalance if max_load.request_count > 0: ratio = min_load.request_count / max_load.request_count if ratio < (1 - self.max_load_diff): logger.info(f"触发负载均衡: {min_load.node_id}({min_load.request_count}) " f"vs {max_load.node_id}({max_load.request_count})") self._move_shard(min_load.node_id, max_load.node_id) def _move_shard(self, from_node: str, to_node: str): """将一个分片从繁忙节点移到空闲节点""" # 获取繁忙节点的一个分片 distribution = self.shard_manager.get_distribution() from_shards = distribution.get(from_node, {}).get('shards', []) if not from_shards: return # 移动第一个分片 shard_id = from_shards[0] logger.info(f"移动分片 {shard_id}: {from_node} → {to_node}") # 这里触发实际的数据迁移 # ... (具体迁移逻辑在后续实现)

五、集成到节点

5.1 增强节点类

# minikv/node_with_sharding.py from typing import List, Optional from .node import MiniKVNode from .hashring.shard_manager import ShardManager from .hashring.load_balancer import LoadBalancer, NodeLoad class MiniKVShardedNode(MiniKVNode): """带分片的 MiniKV 节点""" def __init__(self, node_id: str, host: str, port: int, replication_factor: int = 3): super().__init__(node_id, host, port) # 分片管理 self.shard_manager = ShardManager( local_node_id=node_id, replication_factor=replication_factor ) # 负载均衡 self.load_balancer = LoadBalancer(self.shard_manager) # 本节点存储的数据（按分片组织） self.shard_data: dict = {} def start(self): """启动节点""" super().start() # 将自己加入分片环 self.shard_manager.add_node(self.node_id) # 启动负载均衡 self.load_balancer.start() # 启动负载上报 self._start_load_reporting() def stop(self): """停止节点""" self.load_balancer.stop() super().stop() def on_node_join(self, node_info): """节点加入时更新分片""" super().on_node_join(node_info) self.shard_manager.add_node(node_info.node_id) def on_node_leave(self, node_info): """节点离开时更新分片""" super().on_node_leave(node_info) self.shard_manager.remove_node(node_info.node_id) def put(self, key: str, value: str) -> bool: """ 存储键值对 根据一致性哈希决定存储到哪个节点 """ if self.shard_manager.should_handle_key(key): # 本节点负责 shard = self.shard_manager.get_shard_for_key(key) if shard: if shard.shard_id not in self.shard_data: self.shard_data[shard.shard_id] = {} self.shard_data[shard.shard_id][key] = value return True else: # 转发到目标节点 target = self.shard_manager.hash_ring.get_node(key) if target and target != self.node_id: # 通过网络转发 msg = self._create_kv_message('PUT', key, value) self.transport.send_to(target, msg) return True return False def get(self, key: str) -> Optional[str]: """ 获取键值对 """ if self.shard_manager.should_handle_key(key): shard = self.shard_manager.get_shard_for_key(key) if shard and shard.shard_id in self.shard_data: return self.shard_data[shard.shard_id].get(key) else: target = self.shard_manager.hash_ring.get_node(key) if target and target != self.node_id: msg = self._create_kv_message('GET', key) # 发送请求并等待响应 # ... (异步处理) pass return None def _start_load_reporting(self): """启动负载上报""" import threading import time def report_loop(): while True: time.sleep(5) # 计算本节点负载 load = NodeLoad( node_id=self.node_id, shard_count=len(self.shard_data) ) self.load_balancer.report_load(self.node_id, load) thread = threading.Thread(target=report_loop, daemon=True) thread.start() def _create_kv_message(self, operation: str, key: str, value: str = None): """创建 KV 操作消息""" from .transport.message import Message, MessageType msg_type_map = { 'GET': MessageType.GET, 'PUT': MessageType.PUT, 'DELETE': MessageType.DELETE, } body = {'key': key} if value is not None: body['value'] = value return Message( msg_type=msg_type_map.get(operation, MessageType.GET), sender_id=self.node_id, body=body )

六、完整演示

# examples/sharding_demo.py import time import logging import sys logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(name)s: %(message)s' ) sys.path.insert(0, '..') from minikv.hashring.consistent_hash import HashRing from minikv.hashring.replicated_hash import ReplicatedHashRing from minikv.hashring.shard_manager import ShardManager def demo_hash_ring(): """演示一致性哈希环""" print("=" * 60) print("🎯 一致性哈希环演示") print("=" * 60) # 创建哈希环 ring = HashRing( nodes=['node-A', 'node-B', 'node-C'], virtual_nodes=10 # 为了演示效果，使用较少虚拟节点 ) print("\n📊 初始节点分布:") load = ring.get_node_load() for node, ratio in sorted(load.items()): bar = '█' * int(ratio * 50) print(f" {node}: {bar} {ratio:.1%}") # 测试 key 分布 print("\n🔑 Key 分布测试:") test_keys = [f"user:{i}" for i in range(20)] distribution = {} for key in test_keys: node = ring.get_node(key) distribution[node] = distribution.get(node, 0) + 1 for node, count in sorted(distribution.items()): bar = '█' * count print(f" {node}: {bar} ({count} keys)") # 添加新节点 print("\n➕ 添加 node-D:") ring.add_node('node-D') # 检查哪些 key 迁移了 migrated = 0 for key in test_keys: new_node = ring.get_node(key) old_node = distribution.get(key) if new_node != old_node: migrated += 1 print(f" 迁移的 key 数: {migrated}/{len(test_keys)}") print(f" (传统取模会迁移 {len(test_keys)} 个)") # 移除节点 print("\n➖ 移除 node-B:") ring.remove_node('node-B') print(f" 剩余节点: {ring.get_nodes_count()}") def demo_replication(): """演示副本放置""" print("\n" + "=" * 60) print("📋 副本放置演示") print("=" * 60) ring = ReplicatedHashRing( nodes=['node-1', 'node-2', 'node-3', 'node-4', 'node-5'], replication_factor=3, virtual_nodes=100 ) test_keys = ['user:1001', 'order:2024001', 'product:X-200'] print("\nKey 的副本分布:") for key in test_keys: replicas = ring.get_replica_nodes(key) primary = ring.get_node(key) print(f"\n 📌 {key}:") print(f" 主节点: {primary}") print(f" 副本: {replicas}") def demo_shard_manager(): """演示分片管理器""" print("\n" + "=" * 60) print("🗂️ 分片管理器演示") print("=" * 60) manager = ShardManager( local_node_id='node-1', replication_factor=2 ) # 添加节点 print("\n📡 添加节点:") for i in range(1, 4): manager.add_node(f'node-{i}') print(f" 添加 node-{i}") # 查看分布 print("\n📊 数据分布:") dist = manager.get_distribution() for node, info in dist.items(): print(f" {node}: 负载={info['load']:.1%}") # 模拟节点故障 print("\n💥 模拟 node-2 故障:") manager.remove_node('node-2') print("\n📊 故障后分布:") dist = manager.get_distribution() for node, info in dist.items(): print(f" {node}: 负载={info['load']:.1%}") if __name__ == "__main__": demo_hash_ring() demo_replication() demo_shard_manager()

七、测试

# tests/test_hashring.py import unittest from minikv.hashring.consistent_hash import HashRing from minikv.hashring.replicated_hash import ReplicatedHashRing class TestHashRing(unittest.TestCase): """一致性哈希测试""" def setUp(self): self.ring = HashRing( nodes=['A', 'B', 'C'], virtual_nodes=1000 ) def test_key_distribution(self): """测试 key 分布均匀性""" keys = [f"key:{i}" for i in range(10000)] distribution = {} for key in keys: node = self.ring.get_node(key) distribution[node] = distribution.get(node, 0) + 1 # 检查分布是否相对均匀（最大偏差不超过 20%） counts = list(distribution.values()) avg = sum(counts) / len(counts) max_deviation = max(abs(c - avg) / avg for c in counts) self.assertLess(max_deviation, 0.2) def test_minimal_migration(self): """测试节点增减时迁移最小化""" keys = [f"key:{i}" for i in range(1000)] # 记录原始分布 original = {key: self.ring.get_node(key) for key in keys} # 添加节点 self.ring.add_node('D') # 统计迁移 migrated = sum(1 for key in keys if self.ring.get_node(key) != original[key]) # 迁移量应小于 1/N（约 25%） self.assertLess(migrated / len(keys), 0.3) def test_consistency(self): """测试一致性：同一个 key 始终映射到同一个节点""" key = "test_key" first = self.ring.get_node(key) for _ in range(100): self.assertEqual(self.ring.get_node(key), first) def test_empty_ring(self): """测试空环""" empty_ring = HashRing() self.assertIsNone(empty_ring.get_node("key")) self.assertEqual(empty_ring.get_nodes_count(), 0) def test_node_weight(self): """测试节点权重""" ring = HashRing(virtual_nodes=100) ring.add_node('heavy', weight=3) ring.add_node('light', weight=1) load = ring.get_node_load() heavy_load = load.get('heavy', 0) light_load = load.get('light', 0) # 重节点的负载应该是轻节点的约 3 倍 self.assertAlmostEqual(heavy_load / light_load, 3.0, delta=0.5) class TestReplicatedHashRing(unittest.TestCase): """副本哈希环测试""" def setUp(self): self.ring = ReplicatedHashRing( nodes=['A', 'B', 'C', 'D', 'E'], replication_factor=3 ) def test_replication_factor(self): """测试副本数""" replicas = self.ring.get_replica_nodes("some_key") self.assertEqual(len(replicas), 3) self.assertEqual(len(set(replicas)), 3) # 不能有重复节点 def test_primary_is_in_replicas(self): """测试主节点在副本列表中""" key = "test_key" primary = self.ring.get_node(key) replicas = self.ring.get_replica_nodes(key) self.assertIn(primary, replicas) if __name__ == "__main__": unittest.main()

八、总结

这一讲我们实现了一致性哈希：

关键成果：

• 节点增减时，仅迁移 1/N 的数据（传统取模是全部迁移）

• 通过虚拟节点解决了数据倾斜问题

• 支持权重，不同性能的节点承担不同负载

• 实现了多副本，为后续的容错打下基础

下一讲：我们将实现分布式系统的核心——Raft 共识算法，让集群在节点故障时仍能保持一致。

🧰开发之余的小工具推荐​

处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求，我常用一个纯前端本地工具箱：zz365.top（子页 PDF 大师：PDF 大师 - zz365工具箱）。所有计算在浏览器完成，文件不上传服务器，关页即清。免费、无登录、无广告，适合开发者当常驻标签页。