散列表的核心概念与原理解析

class SimpleHashTable:
    """基础散列表实现"""
    def __init__(self, size=16):
        self.size = size
        self.buckets = [None] * size
        self.count = 0
    
    def _hash(self, key):
        """简单的哈希函数"""
        if isinstance(key, str):
            return sum(ord(c) for c in key) % self.size
        elif isinstance(key, int):
            return key % self.size
        else:
            return hash(key) % self.size
    
    def put(self, key, value):
        """插入键值对"""
        index = self._hash(key)
        self.buckets[index] = value
        self.count += 1
    
    def get(self, key):
        """获取值"""
        index = self._hash(key)
        return self.buckets[index]

class AdvancedHashFunctions:
    """高级哈希函数集合"""
    
    @staticmethod
    def djb2_hash(key):
        """DJB2哈希算法 - 经典字符串哈希"""
        hash_value = 5381
        for char in str(key):
            hash_value = ((hash_value << 5) + hash_value) + ord(char)
        return hash_value & 0xFFFFFFFF
    
    @staticmethod
    def sdbm_hash(key):
        """SDBM哈希算法"""
        hash_value = 0
        for char in str(key):
            hash_value = ord(char) + (hash_value << 6) + (hash_value << 16) - hash_value
        return hash_value & 0xFFFFFFFF
    
    @staticmethod
    def fnv_hash(key):
        """FNV哈希算法"""
        fnv_prime = 16777619
        hash_value = 2166136261
        
        for char in str(key):
            hash_value = hash_value ^ ord(char)
            hash_value = (hash_value * fnv_prime) & 0xFFFFFFFF
        
        return hash_value
 
# 性能测试
import time
 
def benchmark_hash_functions():
    """哈希函数性能对比"""
    test_data = ["hello_world", "python_programming", "hash_table_example"] * 1000
    
    hash_funcs = {
        "DJB2": AdvancedHashFunctions.djb2_hash,
        "SDBM": AdvancedHashFunctions.sdbm_hash,
        "FNV": AdvancedHashFunctions.fnv_hash,
        "Python内置": hash
    }
    
    results = {}
    for name, func in hash_funcs.items():
        start_time = time.time()
        [func(data) for data in test_data]
        end_time = time.time()
        results[name] = end_time - start_time
    
    return results
 
# 执行性能测试
performance_results = benchmark_hash_functions()
for name, time_taken in sorted(performance_results.items(), key=lambda x: x[1]):
    print(f"{name}: {time_taken:.4f}秒")

class ChainedHashTable:
    """链地址法实现的散列表"""
    
    class Node:
        def __init__(self, key, value):
            self.key = key
            self.value = value
            self.next = None
    
    def __init__(self, size=16):
        self.size = size
        self.buckets = [None] * size
        self.count = 0
    
    def _hash(self, key):
        return AdvancedHashFunctions.djb2_hash(key) % self.size
    
    def put(self, key, value):
        """插入键值对 - 处理冲突"""
        index = self._hash(key)
        
        # 检查是否已存在该key
        current = self.buckets[index]
        while current:
            if current.key == key:
                current.value = value  # 更新现有值
                return
            current = current.next
        
        # 插入新节点
        new_node = self.Node(key, value)
        new_node.next = self.buckets[index]
        self.buckets[index] = new_node
        self.count += 1
        
        # 负载因子检查
        if self.count / self.size > 0.75:
            self._resize()
    
    def get(self, key):
        """获取值"""
        index = self._hash(key)
        current = self.buckets[index]
        
        while current:
            if current.key == key:
                return current.value
            current = current.next
        
        return None  # Key不存在
    
    def remove(self, key):
        """删除键值对"""
        index = self._hash(key)
        current = self.buckets[index]
        prev = None
        
        while current:
            if current.key == key:
                if prev:
                    prev.next = current.next
                else:
                    self.buckets[index] = current.next
                self.count -= 1
                return True
            prev = current
            current = current.next
        
        return False
    
    def _resize(self):
        """动态扩容"""
        old_buckets = self.buckets
        self.size *= 2
        self.buckets = [None] * self.size
        self.count = 0
        
        # 重新哈希所有元素
        for head in old_buckets:
            current = head
            while current:
                self.put(current.key, current.value)
                current = current.next

class OpenAddressingHashTable:
    """开放地址法实现的散列表"""
    
    def __init__(self, size=16):
        self.size = size
        self.keys = [None] * size
        self.values = [None] * size
        self.deleted = [False] * size  # 标记已删除的位置
        self.count = 0
    
    def _hash(self, key):
        return AdvancedHashFunctions.fnv_hash(key) % self.size
    
    def _probe(self, key, i):
        """线性探测"""
        return (self._hash(key) + i) % self.size
    
    def _find_slot(self, key, for_insert=False):
        """查找合适的槽位"""
        i = 0
        first_deleted = None
        
        while i < self.size:
            index = self._probe(key, i)
            
            if self.keys[index] is None and not self.deleted[index]:
                # 找到空槽
                return (index, False) if for_insert else (None, False)
            
            if self.deleted[index] and first_deleted is None:
                first_deleted = index
            
            if self.keys[index] == key and not self.deleted[index]:
                return (index, True)  # 找到key
            
            i += 1
        
        # 表已满或找不到key
        return (first_deleted, False) if for_insert and first_deleted is not None else (None, False)
    
    def put(self, key, value):
        """插入键值对"""
        if self.count / self.size > 0.7:
            self._resize()
        
        index, found = self._find_slot(key, for_insert=True)
        
        if index is None:
            raise Exception("哈希表已满")
        
        self.keys[index] = key
        self.values[index] = value
        self.deleted[index] = False
        
        if not found:
            self.count += 1
    
    def get(self, key):
        """获取值"""
        index, found = self._find_slot(key)
        return self.values[index] if found else None
    
    def remove(self, key):
        """删除键值对"""
        index, found = self._find_slot(key)
        if found:
            self.deleted[index] = True
            self.keys[index] = None
            self.values[index] = None
            self.count -= 1
            return True
        return False

class OptimizedHashTable:
    """优化的散列表实现"""
    
    def __init__(self, initial_capacity=16, load_factor=0.75):
        self.capacity = initial_capacity
        self.load_factor = load_factor
        self.threshold = int(self.capacity * self.load_factor)
        self.size = 0
        
        # 使用更好的数据结构
        self.buckets = [[] for _ in range(self.capacity)]
    
    def _hash(self, key):
        """改进的哈希函数"""
        h = 0
        for char in str(key):
            h = (h << 5) - h + ord(char)
            h = h & h  # 保持32位
        return h % self.capacity
    
    def put(self, key, value):
        """优化的插入操作"""
        if self.size >= self.threshold:
            self._resize()
        
        index = self._hash(key)
        bucket = self.buckets[index]
        
        # 检查是否已存在
        for i, (k, v) in enumerate(bucket):
            if k == key:
                bucket[i] = (key, value)  # 更新
                return
        
        # 添加新元素
        bucket.append((key, value))
        self.size += 1
    
    def _resize(self):
        """智能扩容"""
        old_buckets = self.buckets
        self.capacity *= 2
        self.threshold = int(self.capacity * self.load_factor)
        self.size = 0
        
        self.buckets = [[] for _ in range(self.capacity)]
        
        # 重新分布所有元素
        for bucket in old_buckets:
            for key, value in bucket:
                self.put(key, value)
 
# 性能测试函数
def benchmark_hash_table_operations():
    """性能基准测试"""
    import random
    import string
    
    # 生成测试数据
    test_keys = [''.join(random.choices(string.ascii_letters, k=10)) for _ in range(10000)]
    test_values = list(range(10000))
    
    # 测试不同实现
    implementations = {
        "基础实现": ChainedHashTable(1000),
        "优化实现": OptimizedHashTable(1000)
    }
    
    results = {}
    
    for name, hash_table in implementations.items():
        # 插入性能测试
        start_time = time.time()
        for key, value in zip(test_keys, test_values):
            hash_table.put(key, value)
        insert_time = time.time() - start_time
        
        # 查找性能测试
        start_time = time.time()
        for key in test_keys[:1000]:  # 测试部分数据
            hash_table.get(key)
        search_time = time.time() - start_time
        
        results[name] = {
            "插入时间": insert_time,
            "查找时间": search_time
        }
    
    return results
 
# 运行性能测试
perf_results = benchmark_hash_table_operations()
for impl, metrics in perf_results.items():
    print(f"\n{impl}:")
    for metric, value in metrics.items():
        print(f"  {metric}: {value:.4f}秒")

class LRUCache:
    """基于散列表和双向链表的LRU缓存"""
    
    class Node:
        def __init__(self, key, value):
            self.key = key
            self.value = value
            self.prev = None
            self.next = None
    
    def __init__(self, capacity):
        self.capacity = capacity
        self.cache = {}  # 散列表用于快速查找
        self.head = self.Node(0, 0)  # 伪头部
        self.tail = self.Node(0, 0)  # 伪尾部
        self.head.next = self.tail
        self.tail.prev = self.head
    
    def _remove(self, node):
        """从链表中移除节点"""
        prev_node = node.prev
        next_node = node.next
        prev_node.next = next_node
        next_node.prev = prev_node
    
    def _add_to_head(self, node):
        """将节点添加到头部"""
        node.prev = self.head
        node.next = self.head.next
        self.head.next.prev = node
        self.head.next = node
    
    def get(self, key):
        """获取缓存值"""
        if key in self.cache:
            node = self.cache[key]
            self._remove(node)
            self._add_to_head(node)
            return node.value
        return -1
    
    def put(self, key, value):
        """添加缓存"""
        if key in self.cache:
            self._remove(self.cache[key])
        
        new_node = self.Node(key, value)
        self.cache[key] = new_node
        self._add_to_head(new_node)
        
        if len(self.cache) > self.capacity:
            # 移除最久未使用的节点
            lru_node = self.tail.prev
            self._remove(lru_node)
            del self.cache[lru_node.key]
 
# 使用示例
lru = LRUCache(3)
lru.put(1, "用户A的信息")
lru.put(2, "用户B的信息")
lru.put(3, "用户C的信息")
 
print(lru.get(1))  # 输出: 用户A的信息
lru.put(4, "用户D的信息")  # 移除用户B的信息
print(lru.get(2))  # 输出: -1 (已移除)

import bisect
import hashlib
 
class ConsistentHash:
    """一致性哈希实现"""
    
    def __init__(self, replicas=3):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
    
    def _hash(self, key):
        """使用MD5哈希"""
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node):
        """添加节点"""
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_value = self._hash(virtual_key)
            self.ring[hash_value] = node
            bisect.insort(self.sorted_keys, hash_value)
    
    def remove_node(self, node):
        """移除节点"""
        for i in range(self.replicas):
            virtual_key = f"{node}:{i}"
            hash_value = self._hash(virtual_key)
            del self.ring[hash_value]
            self.sorted_keys.remove(hash_value)
    
    def get_node(self, key):
        """获取负责该key的节点"""
        if not self.ring:
            return None
        
        hash_value = self._hash(key)
        
        # 找到第一个大于等于hash_value的位置
        index = bisect.bisect_left(self.sorted_keys, hash_value)
        
        # 如果超出范围，回到第一个节点
        if index == len(self.sorted_keys):
            index = 0
        
        return self.ring[self.sorted_keys[index]]
 
# 使用示例
consistent_hash = ConsistentHash(replicas=3)
consistent_hash.add_node("服务器A")
consistent_hash.add_node("服务器B")
consistent_hash.add_node("服务器C")
 
# 测试数据分布
test_keys = [f"用户{i}" for i in range(100)]
node_distribution = {}
 
for key in test_keys:
    node = consistent_hash.get_node(key)
    node_distribution[node] = node_distribution.get(node, 0) + 1
 
print("数据分布:", node_distribution)

# TRAE IDE 智能生成的哈希表模板
def generate_hash_table_template():
    """
    TRAE IDE 根据需求自动生成的哈希表模板
    包含完整的错误处理和性能优化
    """
    template = '''
class OptimizedHashTable:
    """TRAE IDE 优化的散列表实现"""
    
    def __init__(self, initial_capacity=16):
        # 建议的初始容量和负载因子
        self.capacity = self._next_power_of_two(initial_capacity)
        self.load_factor = 0.75
        self.threshold = int(self.capacity * self.load_factor)
        self.size = 0
        
        # 使用更好的哈希函数
        self.hash_func = self._murmur_hash
        
        # 初始化桶数组
        self.buckets = [None] * self.capacity
    
    def _murmur_hash(self, key):
        """MurmurHash3 实现 - 高性能哈希函数"""
        # TRAE IDE 自动插入优化的哈希算法
        data = str(key).encode()
        length = len(data)
        seed = 0x9747b28c
        
        # MurmurHash3 32-bit implementation
        c1 = 0xcc9e2d51
        c2 = 0x1b873593
        
        h1 = seed
        rounded_end = (length & 0xfffffffc)
        
        for i in range(0, rounded_end, 4):
            k1 = (data[i] & 0xff) | ((data[i + 1] & 0xff) << 8) | \
                 ((data[i + 2] & 0xff) << 16) | ((data[i + 3] & 0xff) << 24)
            
            k1 *= c1
            k1 = (k1 << 15) | (k1 >> 17)
            k1 *= c2
            
            h1 ^= k1
            h1 = (h1 << 13) | (h1 >> 19)
            h1 = h1 * 5 + 0xe6546b64
        
        # 处理剩余字节
        k1 = 0
        val = length & 0x03
        
        if val == 3:
            k1 = (data[rounded_end + 2] & 0xff) << 16
        if val in [2, 3]:
            k1 |= (data[rounded_end + 1] & 0xff) << 8
        if val in [1, 2, 3]:
            k1 |= data[rounded_end] & 0xff
            k1 *= c1
            k1 = (k1 << 15) | (k1 >> 17)
            k1 *= c2
            h1 ^= k1
        
        # 最终化
        h1 ^= length
        h1 ^= (h1 >> 16)
        h1 *= 0x85ebca6b
        h1 ^= (h1 >> 13)
        h1 *= 0xc2b2ae35
        h1 ^= (h1 >> 16)
        
        return h1
    
    def _next_power_of_two(self, n):
        """找到大于等于n的最小2的幂"""
        if n <= 1:
            return 1
        return 1 << (n - 1).bit_length()
    
    # 其他方法实现...
    '''
    return template
 
# 使用TRAE IDE生成的模板
exec(generate_hash_table_template())

class TRAEPerformanceMonitor:
    """TRAE IDE 性能监控器"""
    
    def __init__(self):
        self.metrics = {
            'collisions': 0,
            'resize_count': 0,
            'avg_lookup_time': 0,
            'memory_usage': 0
        }
    
    def record_collision(self):
        """记录哈希冲突"""
        self.metrics['collisions'] += 1
        # TRAE IDE 会在冲突率过高时发出警告
        if self.metrics['collisions'] % 100 == 0:
            self._alert_high_collision_rate()
    
    def _alert_high_collision_rate(self):
        """高冲突率警告"""
        print("⚠️ TRAE IDE 提示：哈希冲突率较高，建议优化哈希函数或扩容")
        # TRAE IDE 会自动推荐优化方案
 
# 集成到哈希表实现中
class TRAEMonitoredHashTable(OptimizedHashTable):
    """集成TRAE监控的散列表"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.monitor = TRAEPerformanceMonitor()
    
    def put(self, key, value):
        """带监控的插入操作"""
        # TRAE IDE 性能分析开始
        start_time = time.perf_counter()
        
        index = self._hash(key) % self.capacity
        if self.buckets[index] is not None:
            self.monitor.record_collision()
        
        super().put(key, value)
        
        # TRAE IDE 性能分析结束
        end_time = time.perf_counter()
        operation_time = end_time - start_time
        
        # 实时性能反馈
        if operation_time > 0.001:  # 1毫秒阈值
            print(f"TRAE IDE: 插入操作耗时{operation_time*1000:.2f}ms，建议检查性能")

def visualize_hash_table(hash_table):
    """
    TRAE IDE 散列表可视化函数
    在调试器中直观展示哈希表结构
    """
    print("\n" + "="*50)
    print("TRAE IDE 散列表可视化")
    print("="*50)
    
    for i, bucket in enumerate(hash_table.buckets):
        if bucket is not None:
            print(f"桶[{i:2d}]: {bucket}")
        else:
            print(f"桶[{i:2d}]: 空")
    
    print(f"\n统计信息:")
    print(f"- 总容量: {hash_table.capacity}")
    print(f"- 已使用: {hash_table.size}")
    print(f"- 负载因子: {hash_table.size / hash_table.capacity:.2f}")
 
# 在TRAE IDE中设置断点进行可视化调试
if __debug__:
    # 创建测试哈希表
    test_ht = TRAEMonitoredHashTable()
    
    # 添加测试数据
    test_data = {
        "user:123": "张三",
        "user:456": "李四",
        "product:789": "iPhone",
        "order:101": "订单信息"
    }
    
    for key, value in test_data.items():
        test_ht.put(key, value)
    
    # TRAE IDE 断点位置 - 可视化哈希表状态
    visualize_hash_table(test_ht)

def select_hash_table_implementation(use_case):
    """
    根据使用场景选择最优的哈希表实现
    TRAE IDE 智能推荐系统
    """
    recommendations = {
        "内存敏感": {
            "implementation": "OpenAddressingHashTable",
            "reason": "开放地址法内存利用率更高，无链表指针开销",
            "config": {"initial_size": 32, "load_factor": 0.6}
        },
        "高并发": {
            "implementation": "ConcurrentHashTable",
            "reason": "需要分段锁或其他并发控制机制",
            "config": {"segments": 16, "concurrency_level": 8}
        },
        "大数据量": {
            "implementation": "ChainedHashTable",
            "reason": "链地址法在扩容时表现更稳定",
            "config": {"initial_size": 1024, "load_factor": 0.75}
        },
        "频繁删除": {
            "implementation": "ChainedHashTable",
            "reason": "链地址法删除操作更简单高效",
            "config": {"initial_size": 64, "load_factor": 0.8}
        }
    }
    
    return recommendations.get(use_case, recommendations["大数据量"])
 
# TRAE IDE 使用建议
print("TRAE IDE 推荐配置:")
config = select_hash_table_implementation("高并发")
print(f"实现方式: {config['implementation']}")
print(f"推荐理由: {config['reason']}")
print(f"配置参数: {config['config']}")

class HashTableMetrics:
    """哈希表性能指标收集"""
    
    def __init__(self):
        self.operations = []
        self.collision_history = []
        self.resize_history = []
    
    def record_operation(self, operation_type, key, duration):
        """记录操作性能"""
        self.operations.append({
            'type': operation_type,
            'key_size': len(str(key)),
            'duration': duration,
            'timestamp': time.time()
        })
    
    def get_performance_report(self):
        """生成性能报告"""
        if not self.operations:
            return "无操作记录"
        
        total_ops = len(self.operations)
        avg_duration = sum(op['duration'] for op in self.operations) / total_ops
        
        by_type = {}
        for op in self.operations:
            op_type = op['type']
            if op_type not in by_type:
                by_type[op_type] = []
            by_type[op_type].append(op['duration'])
        
        report = f"""
性能报告 (总操作数: {total_ops})
=====================================
平均响应时间: {avg_duration*1000:.2f}ms
 
各操作类型统计:
"""
        for op_type, durations in by_type.items():
            avg_time = sum(durations) / len(durations)
            report += f"  {op_type}: {avg_time*1000:.2f}ms (调用次数: {len(durations)})\n"
        
        return report
 
# 集成到生产环境代码中
class ProductionHashTable:
    """生产环境哈希表实现"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = HashTableMetrics()
    
    def put(self, key, value):
        """带性能监控的插入"""
        start_time = time.perf_counter()
        result = super().put(key, value)
        duration = time.perf_counter() - start_time
        
        self.metrics.record_operation('PUT', key, duration)
        return result
    
    def get(self, key):
        """带性能监控的查询"""
        start_time = time.perf_counter()
        result = super().get(key)
        duration = time.perf_counter() - start_time
        
        self.metrics.record_operation('GET', key, duration)
        return result