上一讲我们实现了节点发现和网络通信三个节点能互相感知了。但有个核心问题没解决数据该存到哪个节点上最简单的做法是取模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 虚拟节点为了解决两个问题数据倾斜节点在环上分布不均匀导致某些节点数据过多负载不均不同节点性能不同需要差异化权重引入虚拟节点每个物理节点对应多个虚拟节点分散在环上。物理节点 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 fHashRing(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_factorreplication_factor, virtual_nodesvirtual_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(targetself._balance_loop, daemonTrue) 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(keylambda 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}) fvs {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_idnode_id, replication_factorreplication_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_idself.node_id, shard_countlen(self.shard_data) ) self.load_balancer.report_load(self.node_id, load) thread threading.Thread(targetreport_loop, daemonTrue) 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_typemsg_type_map.get(operation, MessageType.GET), sender_idself.node_id, bodybody )六、完整演示# examples/sharding_demo.py import time import logging import sys logging.basicConfig( levellogging.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_nodes10 # 为了演示效果使用较少虚拟节点 ) 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 [fuser:{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_factor3, virtual_nodes100 ) 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_idnode-1, replication_factor2 ) # 添加节点 print(\n 添加节点:) for i in range(1, 4): manager.add_node(fnode-{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_nodes1000 ) def test_key_distribution(self): 测试 key 分布均匀性 keys [fkey:{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 [fkey:{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_nodes100) ring.add_node(heavy, weight3) ring.add_node(light, weight1) 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, delta0.5) class TestReplicatedHashRing(unittest.TestCase): 副本哈希环测试 def setUp(self): self.ring ReplicatedHashRing( nodes[A, B, C, D, E], replication_factor3 ) 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工具箱。所有计算在浏览器完成文件不上传服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。