15.2 布隆过滤器

前置：哈希表与位运算。布隆过滤器回答的是概率成员关系：对阴性结果可确定“未加入”，对阳性结果只能判断“可能已加入”。

名单太大，答案可以有一点模糊

假设后端需要先拦截明显不存在的 key，再把“可能存在”的少数结果送到权威数据库确认。完整保存所有 key 能得到精确答案，但会占用更多空间；Bloom filter 选择了另一个合同。

布隆过滤器用 m 个 bit 和 k 个哈希位置保存集合摘要。加入元素时把对应位置设为 1；查询时只要看到一个 0，就能确定元素从未按同一规则加入。所有位置都是 1 时，可能是元素本身留下的，也可能是其他元素碰巧覆盖了这些位置。

因此，标准布隆过滤器有两条不对称的合同：

• 不会产生假阴性，前提是插入已完整生效、没有丢失或损坏位更新，编码和哈希配置保持一致；
• 允许假阳性，调用方必须准备二次确认。

bit 数组如何表示集合摘要

过滤器不保存完整 key，只把每个 key 映射到 bit 数组中的若干位置。查询时只要有一位仍为 0，就能确定该 key 没有按同一配置加入；全部为 1 时，仍要查询权威数据源。下面的实现使用真实 bit，而不是用 set 偷换空间模型。

下面使用 bytearray，每个字节保存 8 个位置。接口只接受 bytes，让字符编码由调用方决定。本实现要求位数 m 为正的 2 的幂、哈希数在 1 到 m 之间，并拒绝布尔配置；普通布隆过滤器本身没有这个位数限制。这样做是为了让双重哈希的奇数步长与 m 互质，前 k 个位置不会提前循环。跨服务使用时还要固定摘要算法、字节序、位数、哈希数量和序列化版本；对象创建后不修改配置。

import hashlib

class BloomFilter:
    def __init__(self, bit_count, hash_count):
        if (not isinstance(bit_count, int) or isinstance(bit_count, bool)
                or bit_count <= 0 or bit_count & (bit_count - 1)):
            raise ValueError("bit_count must be a positive power of two")
        if (not isinstance(hash_count, int) or isinstance(hash_count, bool)
                or not 1 <= hash_count <= bit_count):
            raise ValueError("hash_count must be an integer in [1, bit_count]")
        self._bit_count = bit_count
        self._hash_count = hash_count
        self._bits = bytearray((bit_count + 7) // 8)

    @property
    def bit_count(self):
        return self._bit_count

    @property
    def hash_count(self):
        return self._hash_count

    def _positions(self, value):
        if not isinstance(value, bytes):
            raise TypeError("BloomFilter values must be bytes")
        digest = hashlib.blake2b(value, digest_size=16).digest()
        first = int.from_bytes(digest[:8], "little")
        second = int.from_bytes(digest[8:], "little") | 1
        for index in range(self.hash_count):
            yield (first + index * second) % self.bit_count

    def add(self, value):
        for position in self._positions(value):
            byte_index, bit_index = divmod(position, 8)
            self._bits[byte_index] |= 1 << bit_index

    def might_contain(self, value):
        for position in self._positions(value):
            byte_index, bit_index = divmod(position, 8)
            if not self._bits[byte_index] & (1 << bit_index):
                return False
        return True

    def byte_size(self):
        return len(self._bits)

if __name__ == "__main__":
    bloom = BloomFilter(bit_count=8_192, hash_count=7)
    inserted = [f"user-{index}".encode() for index in range(500)]
    for item in inserted:
        bloom.add(item)

    assert all(bloom.might_contain(item) for item in inserted)
    assert bloom.byte_size() == 1_024

    probes = [f"outsider-{index}".encode() for index in range(2_000)]
    false_positives = sum(bloom.might_contain(item) for item in probes)
    print("样本误报数：", false_positives)

最后的误报数只是这批探针上的测量，不是概率保证，也不能据一次零误报就宣称误报率为 0。测试应断言所有已插入元素都返回 True，并核对实际位位置、空过滤器和输入错误；统计误报率另作测量。byte_size() 只计位数组的载荷，示例为 1,024 字节，不包含 Python 对象和分配器开销。

bit 数组越满，误报越多

位数 m、已插入的不同元素数 n 和每个元素的位置数 k 共同影响误报。重复加入同一字节串只会再次设置原有位置，不增加这份 n。

在独立均匀哈希模型中，插入 n 个不同元素后，某一位仍为 0 的概率是 (1-1/m)^(kn)，近似为 exp(-kn/m)。再近似处理查询位置的联合事件，可得对未插入元素的误报率：

p ≈ (1 - exp(-k * n / m))^k

给定 m 与预计的 n>0，使误报率较小的哈希数量约为：

k ≈ (m / n) * ln(2)

若给定目标误报率 0<p<1，代入近似最优 k 可反推 m ≈ -n*ln(p)/(ln(2)^2)。实际必须把 m、k 取成合法整数；本实现还要把 m 向上取到 2 的幂，再按实际 m 重新选择 k。这个模型不是对任意输入的硬保证。

k 不是越大越好。哈希位置太少，bit 利用不足；太多则每次操作更贵，也会更快把位数组填满。容量规划应从预计元素数和可接受误报率反推，而不是随手写 k=3。

实际误报还受哈希质量、输入分布和容量超载影响。元素数显著超过设计值后，越来越多 bit 变成 1，过滤器会逐渐失去区分能力。生产系统通常要监控填充率，并安排分代或重建。

为什么不能随手清零一个 bit

两个元素可能映射到同一个 bit。删除其中一个元素时若直接清零该 bit，另一个元素就会被错误判成“不存在”。共享 bit 没有记录所有权。

计数布隆过滤器为每个位置保存计数，可以在限制下支持删除：必须确认这个元素确实被加入过，并且删除次数没有超过加入次数；“查询可能存在”本身不能授权减计数。计数溢出、下溢与并发丢更新同样可能破坏保证，而且计数器占用更多空间。

若需求是精确成员关系、枚举元素或删除后立即严格一致，应使用集合、哈希表或其他索引。布隆过滤器适合放在权威数据源之前减少无效查询，不能替代权威数据。

复杂度与工程边界

操作	时间	空间或误差
加入	O(k)	设置 k 个 bit
查询	O(k)	可能假阳性
存储	O(m) bit	不保存原始元素

摘要计算还要读取 L 个输入字节，因此加入和查询的最坏时间为 O(L+k)，创建对象还需 Θ(ceil(m/8)) 的初始化时间。这里按哈希位置整数运算为常数成本计数。hashlib.blake2b生成 128-bit 摘要，再拆成两个 64-bit 整数做双重哈希；k 个位置不是数学上完全独立的哈希函数。Kirsch 与 Mitzenmacher 的论文分析了双重哈希的渐进误报性质，不能据此把上面的近似式当成本代码的精确概率。

“奇数步长”只在 m 为 2 的幂时保证互质；若随意允许 m=10，步长 5 就只会访问两个位置。这个实现也没有并发锁：读改写同一字节可能丢失其他线程的置位。要用阴性结果跳过数据库，所有相关已提交 key 必须已反映在过滤器中；构建未完成或同步滞后时，应先查询权威数据源。

动手测量过滤器

1. 固定 n 与 m，改变 k，比较实测误报率与近似公式。
2. 让插入量超过设计值两倍，记录 bit 填充率与误报变化。
3. 给配置加入版本号，设计跨进程序列化格式。
4. 解释为什么 might_contain 返回 True 后仍要访问权威存储。
5. 设计一个会因直接清 bit 而制造假阴性的删除例子。

下一步：从概率成员关系到动态连通性

布隆过滤器用允许误报换取紧凑空间。下一篇的并查集不接受概率答案，它维护的是不断合并的连通分量；再下一篇用线段树处理区间聚合与单点更新。