哈希算法题目：设计一个基于哈希的分布式键值存储系统

题目描述
设计一个基于哈希的分布式键值存储系统，需要支持以下操作：

• put(key, value)：存储键值对
• get(key)：根据键获取值，如果键不存在返回特定标识
• remove(key)：删除键值对

系统需要解决以下分布式环境中的关键问题：

1. 数据分片：如何将数据分布到多个节点
2. 一致性哈希：如何实现节点的动态增删而不引起大量数据迁移
3. 容错性：如何处理节点故障
4. 负载均衡：如何保证数据均匀分布

解题过程

第一步：理解基本架构
分布式键值存储系统由多个节点组成，每个节点负责存储一部分数据。我们需要设计一个哈希环来管理这些节点。

关键概念：

• 虚拟节点：每个物理节点对应多个虚拟节点，实现更好的负载均衡
• 一致性哈希：将节点和键映射到同一个哈希环上
• 数据分片：根据键的哈希值决定数据存储在哪个节点

第二步：设计哈希环结构

import hashlib
from bisect import bisect_right

class DistributedHashRing:
    def __init__(self, nodes=None, virtual_nodes=3):
        self.virtual_nodes = virtual_nodes
        self.ring = {}  # 虚拟节点到物理节点的映射
        self.sorted_keys = []  # 排序的虚拟节点哈希值
        
        if nodes:
            for node in nodes:
                self.add_node(node)

第三步：实现节点管理

def _hash(self, key):
    """计算键的哈希值"""
    return int(hashlib.md5(key.encode()).hexdigest(), 16)

def add_node(self, node):
    """添加新节点到哈希环"""
    for i in range(self.virtual_nodes):
        virtual_node = f"{node}:{i}"
        hash_key = self._hash(virtual_node)
        self.ring[hash_key] = node
        self.sorted_keys.append(hash_key)
    
    self.sorted_keys.sort()

def remove_node(self, node):
    """从哈希环移除节点"""
    for i in range(self.virtual_nodes):
        virtual_node = f"{node}:{i}"
        hash_key = self._hash(virtual_node)
        if hash_key in self.ring:
            del self.ring[hash_key]
            self.sorted_keys.remove(hash_key)

第四步：实现数据定位

def get_node(self, key):
    """根据键找到对应的节点"""
    if not self.ring:
        return None
    
    hash_key = self._hash(key)
    idx = bisect_right(self.sorted_keys, hash_key)
    
    # 如果超过最大值，回到环的开头
    if idx == len(self.sorted_keys):
        idx = 0
    
    virtual_node_hash = self.sorted_keys[idx]
    return self.ring[virtual_node_hash]

第五步：设计存储节点类

class StorageNode:
    def __init__(self, node_id):
        self.node_id = node_id
        self.data = {}
        self.replica_nodes = []  # 副本节点
    
    def put(self, key, value):
        self.data[key] = value
    
    def get(self, key):
        return self.data.get(key, None)
    
    def remove(self, key):
        if key in self.data:
            del self.data[key]
            return True
        return False

第六步：实现完整的分布式存储系统

class DistributedKeyValueStore:
    def __init__(self, nodes, replication_factor=2):
        self.hash_ring = DistributedHashRing()
        self.nodes = {}
        self.replication_factor = replication_factor
        
        # 初始化所有节点
        for node_id in nodes:
            self.add_node(node_id)
    
    def add_node(self, node_id):
        """添加新节点"""
        self.hash_ring.add_node(node_id)
        self.nodes[node_id] = StorageNode(node_id)
        
        # 重新平衡数据（简化版）
        self._rebalance_data()
    
    def remove_node(self, node_id):
        """移除节点"""
        if node_id in self.nodes:
            # 先迁移数据
            self._migrate_data_before_removal(node_id)
            self.hash_ring.remove_node(node_id)
            del self.nodes[node_id]
    
    def put(self, key, value):
        """存储键值对"""
        primary_node_id = self.hash_ring.get_node(key)
        if not primary_node_id:
            raise Exception("No available nodes")
        
        # 存储到主节点
        self.nodes[primary_node_id].put(key, value)
        
        # 存储副本
        self._store_replicas(key, value, primary_node_id)
    
    def get(self, key):
        """获取值"""
        node_id = self.hash_ring.get_node(key)
        if node_id and node_id in self.nodes:
            return self.nodes[node_id].get(key)
        return None
    
    def remove(self, key):
        """删除键值对"""
        node_id = self.hash_ring.get_node(key)
        if node_id and node_id in self.nodes:
            # 从主节点删除
            result = self.nodes[node_id].remove(key)
            
            # 从副本删除
            self._remove_replicas(key, node_id)
            return result
        return False

第七步：实现数据复制和容错

def _store_replicas(self, key, value, primary_node_id):
    """存储数据副本"""
    node_ids = self._get_replica_nodes(primary_node_id)
    for node_id in node_ids:
        if node_id in self.nodes:
            self.nodes[node_id].put(key, value)

def _remove_replicas(self, key, primary_node_id):
    """删除数据副本"""
    node_ids = self._get_replica_nodes(primary_node_id)
    for node_id in node_ids:
        if node_id in self.nodes:
            self.nodes[node_id].remove(key)

def _get_replica_nodes(self, primary_node_id):
    """获取副本节点列表"""
    all_nodes = list(self.nodes.keys())
    primary_index = all_nodes.index(primary_node_id)
    
    replica_nodes = []
    for i in range(1, self.replication_factor + 1):
        replica_index = (primary_index + i) % len(all_nodes)
        replica_nodes.append(all_nodes[replica_index])
    
    return replica_nodes

第八步：实现数据迁移（简化版）

def _rebalance_data(self):
    """数据重新平衡（简化实现）"""
    # 在实际系统中，这里需要实现完整的数据迁移逻辑
    pass

def _migrate_data_before_removal(self, node_id):
    """节点移除前的数据迁移"""
    if node_id in self.nodes:
        node_data = self.nodes[node_id].data.copy()
        for key, value in node_data.items():
            # 为每个键找到新的主节点
            new_primary = self.hash_ring.get_node(key)
            if new_primary and new_primary != node_id:
                self.put(key, value)
                self.nodes[node_id].remove(key)

第九步：测试系统功能

# 创建分布式存储系统
nodes = ["node1", "node2", "node3", "node4"]
kv_store = DistributedKeyValueStore(nodes)

# 测试基本操作
kv_store.put("name", "Alice")
kv_store.put("age", "25")
kv_store.put("city", "New York")

print(kv_store.get("name"))  # 输出: Alice
print(kv_store.get("age"))   # 输出: 25

# 测试节点动态添加
kv_store.add_node("node5")
print("After adding node5:")

# 测试节点移除
kv_store.remove_node("node2")
print("After removing node2:")

# 验证数据一致性
print(kv_store.get("name"))  # 应该仍然能获取到数据

关键优化点：

1. 虚拟节点数量：增加虚拟节点数可以提高负载均衡性
2. 数据复制因子：根据可用性要求调整复制因子
3. 一致性哈希算法：选择合适的哈希函数减少冲突
4. 故障检测：需要实现节点健康检查机制
5. 数据迁移策略：实现渐进式数据迁移减少网络压力

这个设计提供了一个基本的分布式键值存储框架，实际生产系统还需要考虑网络通信、持久化存储、事务一致性等更复杂的问题。