哈希算法题目：设计一个基于哈希的分布式事件去重系统（支持时间窗口和滑动过期）

题目描述
设计一个分布式事件去重系统，用于处理来自多个数据源的事件流。每个事件包含唯一标识符、时间戳和有效载荷。系统需要：

1. 在指定时间窗口内对重复事件进行精确去重
2. 支持滑动过期机制，自动清理过期事件
3. 保证高吞吐量和低延迟
4. 支持水平扩展

解题过程

步骤1：系统架构设计
我们采用分片式架构，将事件分散到多个处理节点：

• 使用一致性哈希进行数据分片，保证负载均衡和扩展性
• 每个节点维护本地哈希表和过期清理机制
• 通过哈希函数将事件ID映射到对应节点

步骤2：核心数据结构设计
每个节点维护两个核心数据结构：

class DedupNode:
    def __init__(self, window_size):
        self.event_map = {}  # 存储事件ID与时间戳的映射
        self.expiration_queue = []  # 基于时间排序的优先队列，用于快速过期清理
        self.window_size = window_size  # 去重时间窗口（秒）

步骤3：事件处理流程
当新事件到达时：

1. 提取事件ID和时间戳
2. 计算哈希值确定目标节点：node_index = hash(event_id) % total_nodes
3. 目标节点执行去重检查：

def process_event(self, event_id, timestamp):
    current_time = timestamp
    
    # 步骤3.1：清理过期事件
    self.cleanup_expired(current_time)
    
    # 步骤3.2：检查是否重复
    if event_id in self.event_map:
        existing_time = self.event_map[event_id]
        if current_time - existing_time <= self.window_size:
            return False  # 重复事件，拒绝处理
        else:
            # 更新为更新的时间戳
            self.event_map[event_id] = current_time
            self.update_expiration_queue(event_id, current_time)
            return True
    else:
        # 新事件，添加到系统
        self.event_map[event_id] = current_time
        self.expiration_queue.append((event_id, current_time))
        return True

步骤4：滑动过期机制实现
实现高效的过期事件清理：

def cleanup_expired(self, current_time):
    # 清理所有过期事件（早于时间窗口起点）
    cutoff_time = current_time - self.window_size
    
    # 从过期队列头部开始清理（按时间排序）
    while self.expiration_queue and self.expiration_queue[0][1] <= cutoff_time:
        expired_id, expired_time = self.expiration_queue.pop(0)
        
        # 只有当哈希表中的时间戳匹配时才删除（防止已更新的事件被误删）
        if self.event_map.get(expired_id) == expired_time:
            del self.event_map[expired_id]

步骤5：优化过期队列
使用最小堆提高过期清理效率：

import heapq

def update_expiration_queue(self, event_id, timestamp):
    # 使用最小堆维护过期时间
    heapq.heappush(self.expiration_queue, (timestamp, event_id))

def optimized_cleanup(self, current_time):
    cutoff_time = current_time - self.window_size
    
    while self.expiration_queue and self.expiration_queue[0][0] <= cutoff_time:
        expired_time, expired_id = heapq.heappop(self.expiration_queue)
        
        # 验证事件是否仍然有效且未被更新
        if self.event_map.get(expired_id) == expired_time:
            del self.event_map[expired_id]

步骤6：分布式一致性保证
处理节点故障和数据一致性问题：

• 使用副本机制：每个事件在多个节点备份
• 采用Quorum协议：需要W个节点确认写入，R个节点确认读取
• 实现故障转移：当节点失效时，自动将流量路由到备份节点

步骤7：性能调优策略

1. 批量处理：积累多个清理操作后批量执行
2. 异步清理：在后台线程执行过期清理，不阻塞主流程
3. 内存优化：对事件ID进行压缩存储
4. 监控指标：跟踪内存使用率、处理延迟、去重命中率

这个设计方案能够有效处理大规模事件流的实时去重需求，同时通过滑动过期机制保证内存使用的可控性。