布谷鸟过滤器

基本特点

布谷鸟过滤器（Cuckoo Filter）是一种概率性数据结构，用于高效地判断某个元素是否存在于集合中。它比传统的布隆过滤器（Bloom Filter）更灵活，主要特点包括：

• 支持动态操作：可以插入、查询和删除元素（布隆过滤器不支持删除）。
• 空间效率高：比布隆过滤器占用更少的内存。
• 查询速度快：查询时间复杂度为 O(1)。
• 无假阴性（False Negative）：如果元素存在，查询一定返回”存在”。
• 可能有假阳性（False Positive）：可能误判不存在的元素为”存在”，但概率可控。

---

与布隆过滤器的比较

特性	布谷鸟过滤器	布隆过滤器
支持删除	是	否
空间占用	更低	较高
查询速度	O(1)	O(k)（k 个哈希）
假阳性率 不存在可能被认为存在	可优化	可优化
假阴性率 存在的可能被认为不存在	无	无
实现复杂度	较高	较低
适用场景	动态数据集	静态数据集

具体实现逻辑

核心数据结构

• 桶数组（Buckets）：由多个桶组成，每个桶可存储固定数量的指纹（如 4 个）。
• 指纹（Fingerprint）：元素的哈希摘要（通常 8-16 bits），用于唯一标识元素。
• 两个哈希函数：
  • hash1(x) → 计算第一个候选桶位置 i1
  • hash2(x) = i1 ^ hash(fingerprint(x)) → 计算第二个候选桶位置 i2

插入

1. 计算 x 的指纹 fp = fingerprint(x)。
2. 计算两个候选桶位置：
   • i1 = hash1(x) % bucket_num
   • i2 = i1 ^ hash(fp) % bucket_num
3. 如果 i1 或 i2 有空位，插入 fp 并返回成功。
4. 如果都满，随机踢出 i1 或 i2 中的一个指纹 fp'，并尝试将其重新插入到它的另一个候选位置。
5. 重复步骤 4，直到成功或达到最大踢出次数（如 500 次），此时认为过滤器已满。

存储的时候，存储 fp，而不是将位置标志为 1.

查询

1. 计算 x 的指纹 fp = fingerprint(x)。
2. 计算两个候选桶 i1 和 i2。
3. 检查 i1 和 i2 是否包含 fp：
   • 如果找到 → 返回 “可能存在”（可能有假阳性）。
   • 如果未找到 → 返回 “肯定不存在”（无假阴性）。

删除

1. 计算 x 的指纹 fp = fingerprint(x)。
2. 计算 i1 和 i2。
3. 检查 i1 和 i2，删除匹配的 fp（只删除一个实例）。
4. 如果找不到 fp，说明 x 不存在。

优缺点

优点

1. 支持删除：可以安全地移除元素，而布隆过滤器不行。
2. 更高的空间效率：在相同误判率下，比布隆过滤器占用更少内存。
3. 更快的查询：只需计算 1-2 个哈希，而布隆过滤器需要多个哈希计算。
4. 无假阴性：已存在的元素一定不会被误判为”不存在”。

缺点

1. 实现较复杂：需要处理”踢出”（kick）机制，代码实现比布隆过滤器复杂。
2. 插入可能失败：当过滤器接近满载时，插入可能需要多次重定位，甚至失败。
3. 指纹冲突问题：不同元素可能计算得到相同指纹（但概率很低）。

适用场景

• 数据库加速：快速判断某个键是否存在，避免不必要的磁盘查询。
• 缓存系统：防止缓存穿透（如 Redis 布隆过滤器）。
• 网络去重：检测重复数据包或请求。
• 日志分析：高效统计独立用户或事件。

实现代码示例

import random
 
class CuckooFilter:
    def __init__(self, capacity, bucket_size=4, max_kicks=500):
        """
        初始化布谷鸟过滤器
        
        参数:
            capacity: 过滤器容量 (桶数量)
            bucket_size: 每个桶能存放的指纹数量 (默认4)
            max_kicks: 插入时最大踢出次数 (默认500)
        """
        self.capacity = capacity
        self.bucket_size = bucket_size
        self.max_kicks = max_kicks
        self.buckets = [[] for _ in range(capacity)]  # 初始化空桶
    
    def insert(self, item):
        """
        插入一个元素
        
        返回:
            True 如果插入成功
            False 如果过滤器已满
        """
        fp = self.fingerprint(item)
        i1 = self.hash1(item) % self.capacity
        i2 = self.hash2(fp, i1) % self.capacity
        
        # 尝试直接插入
        if self._insert_into_bucket(i1, fp):
            return True
        if self._insert_into_bucket(i2, fp):
            return True
        
        # 需要踢出操作
        for _ in range(self.max_kicks):
            # 随机选择一个桶踢出
            i = random.choice([i1, i2])
            # 随机选择桶中的一个位置
            if not self.buckets[i]:
                continue  # 空桶，跳过
                
            idx = random.randint(0, len(self.buckets[i])-1)
            victim_fp = self.buckets[i][idx]
            
            # 踢出旧指纹，插入新指纹
            self.buckets[i][idx] = fp
            
            # 计算被踢出指纹的另一个位置
            i = i ^ self.hash2(victim_fp, i) % self.capacity
            
            # 尝试插入被踢出的指纹
            if self._insert_into_bucket(i, victim_fp):
                return True
            
            # 更新要插入的指纹为被踢出的指纹
            fp = victim_fp
        
        # 达到最大踢出次数仍未插入成功
        return False
    
    def lookup(self, item):
        """
        查询元素是否存在
        
        返回:
            True 可能存在 (可能有假阳性)
            False 肯定不存在 (无假阴性)
        """
        fp = self.fingerprint(item)
        i1 = self.hash1(item) % self.capacity
        i2 = self.hash2(fp, i1) % self.capacity
        
        return fp in self.buckets[i1] or fp in self.buckets[i2]
    
    def delete(self, item):
        """
        删除一个元素
        
        返回:
            True 如果删除成功
            False 如果元素不存在
        """
        fp = self.fingerprint(item)
        i1 = self.hash1(item) % self.capacity
        i2 = self.hash2(fp, i1) % self.capacity
        
        # 从第一个桶中删除
        if fp in self.buckets[i1]:
            self.buckets[i1].remove(fp)
            return True
        
        # 从第二个桶中删除
        if fp in self.buckets[i2]:
            self.buckets[i2].remove(fp)
            return True
        
        return False
    
    def _insert_into_bucket(self, index, fp):
        """尝试将指纹插入到指定桶中"""
        if len(self.buckets[index]) < self.bucket_size:
            self.buckets[index].append(fp)
            return True
        return False
    
    # 以下方法需要具体实现
    def fingerprint(self, item):
        """计算元素的指纹 (需实现)"""
        raise NotImplementedError
        
    def hash1(self, item):
        """第一个哈希函数 (需实现)"""
        raise NotImplementedError
        
    def hash2(self, fp, index):
        """第二个哈希函数 (需实现)"""
        raise NotImplementedError