实现一个基于哈希的分布式实时监控系统（支持滑动窗口统计和异常检测）

题目描述
设计一个分布式实时监控系统，用于收集来自多个服务器的指标数据（如CPU使用率、内存使用率等）。系统需要能够：

1. 在滑动时间窗口内（如最近5分钟）统计各项指标
2. 实时检测异常值（如超过阈值的数据点）
3. 支持高并发写入和查询
4. 具备横向扩展能力

核心思路
使用时间分片哈希结合环形缓冲区来管理滑动窗口数据，通过布隆过滤器进行快速异常检测。

详细实现步骤

第一步：设计滑动窗口数据结构

class TimeWindow:
    def __init__(self, window_size=300, precision=1):  # 5分钟窗口，1秒精度
        self.window_size = window_size  # 窗口大小（秒）
        self.precision = precision      # 时间精度（秒）
        self.slots = window_size // precision  # 时间槽数量
        self.ring_buffer = [{} for _ in range(self.slots)]  # 环形数组
        self.current_index = 0
        self.last_update = 0
        
    def _get_slot_index(self, timestamp):
        """将时间戳映射到环形缓冲区索引"""
        slot_time = timestamp // self.precision
        return slot_time % self.slots

第二步：实现数据写入机制

def add_metric(self, metric_name, value, timestamp):
    # 清理过期数据
    self._evict_expired(timestamp)
    
    # 获取当前时间槽索引
    slot_index = self._get_slot_index(timestamp)
    
    # 如果切换到新时间槽，清空旧数据
    if timestamp // self.precision > self.last_update:
        self.ring_buffer[slot_index] = {}
        self.last_update = timestamp // self.precision
    
    # 更新统计信息
    if metric_name not in self.ring_buffer[slot_index]:
        self.ring_buffer[slot_index][metric_name] = {
            'count': 0, 'sum': 0, 'min': float('inf'), 'max': float('-inf')
        }
    
    stats = self.ring_buffer[slot_index][metric_name]
    stats['count'] += 1
    stats['sum'] += value
    stats['min'] = min(stats['min'], value)
    stats['max'] = max(stats['max'], value)

第三步：实现滑动窗口查询

def get_window_stats(self, metric_name, current_time):
    """获取指定指标在滑动窗口内的统计信息"""
    total_count = 0
    total_sum = 0
    global_min = float('inf')
    global_max = float('-inf')
    
    # 遍历所有有效的时间槽
    for i in range(self.slots):
        slot_time = (current_time // self.precision) - (self.slots - 1) + i
        if slot_time < self.last_update - self.slots + 1:
            continue  # 跳过过期槽
            
        slot_index = slot_time % self.slots
        if metric_name in self.ring_buffer[slot_index]:
            stats = self.ring_buffer[slot_index][metric_name]
            total_count += stats['count']
            total_sum += stats['sum']
            global_min = min(global_min, stats['min'])
            global_max = max(global_max, stats['max'])
    
    return {
        'count': total_count,
        'avg': total_sum / total_count if total_count > 0 else 0,
        'min': global_min,
        'max': global_max
    }

第四步：实现基于布隆过滤器的异常检测

class AnomalyDetector:
    def __init__(self, capacity=100000, error_rate=0.01):
        from pybloom_live import ScalableBloomFilter
        self.normal_values = ScalableBloomFilter(capacity=capacity, error_rate=error_rate)
        self.threshold = 3.0  # 3个标准差
        
    def is_anomaly(self, metric_name, value, historical_data):
        """检测是否为异常值"""
        # 计算历史数据的均值和标准差
        if len(historical_data) < 10:  # 需要有足够的历史数据
            return False
            
        mean = sum(historical_data) / len(historical_data)
        variance = sum((x - mean) ** 2 for x in historical_data) / len(historical_data)
        std_dev = variance ** 0.5
        
        # 使用Z-score检测异常
        z_score = abs(value - mean) / std_dev if std_dev > 0 else 0
        return z_score > self.threshold

第五步：分布式架构设计

class DistributedMonitor:
    def __init__(self, num_shards=10):
        self.shards = [TimeWindow() for _ in range(num_shards)]
        self.detectors = [AnomalyDetector() for _ in range(num_shards)]
        
    def _get_shard(self, metric_name):
        """基于指标名称的哈希值进行分片"""
        hash_val = hash(metric_name)
        return hash_val % len(self.shards)
    
    def report_metric(self, metric_name, value, timestamp):
        shard_index = self._get_shard(metric_name)
        shard = self.shards[shard_index]
        detector = self.detectors[shard_index]
        
        # 添加到时间窗口
        shard.add_metric(metric_name, value, timestamp)
        
        # 异常检测
        historical_data = self._get_historical_data(metric_name, timestamp)
        if detector.is_anomaly(metric_name, value, historical_data):
            self._trigger_alert(metric_name, value, timestamp)

关键要点

1. 时间分片哈希确保数据均匀分布
2. 环形缓冲区实现高效的内存使用
3. 布隆过滤器提供快速的异常检测
4. 分片设计支持水平扩展
5. 滑动窗口保证实时性同时控制数据量

这个设计能够处理高频率的指标上报，在有限内存下提供准确的滑动窗口统计，并快速识别异常模式。