用 Python 位数组实现布隆过滤器拦截不存在的短链 ID

问题场景

短链服务通常用短链 ID 查询映射关系。若请求携带大量不存在的 ID，缓存未命中后就会直接查数据库，形成缓存穿透。布隆过滤器可以在进入缓存和数据库前做一次前置判断：某个 ID 一定不存在，或可能存在。对“一定不存在”的 ID 直接返回，就能挡住一部分无效请求。

方案与参数

布隆过滤器使用位数组和 k 个哈希函数。插入时把 k 个位置置 1；查询时若任一位置为 0，则一定不存在；若全为 1，则可能存在。误判率 p 与位数组大小 m、哈希次数 k 有关。按预期元素数 n 和可接受误判率 p 计算：

m = ceil(-n * ln(p) / (ln(2) ** 2)) k = max(1, round((m / n) * ln(2)))

双哈希可以用一次哈希生成两个整数 h1、h2，第 i 个位置为 (h1 + i * h2) % m。示例使用 hashlib.blake2b，避免 Python 内置 hash 的随机化导致进程重启后结果不一致。

最小可运行代码

import math
import hashlib


class BloomFilter:
    def __init__(self, expected_items, false_positive_rate):
        if expected_items <= 0:
            raise ValueError('expected_items 必须大于 0')
        if not (0 < false_positive_rate < 1):
            raise ValueError('false_positive_rate 必须在 0 和 1 之间')

        n = expected_items
        p = false_positive_rate
        self.m = max(1, math.ceil(-n * math.log(p) / (math.log(2) ** 2)))
        self.k = max(1, round((self.m / n) * math.log(2)))
        self.bits = bytearray((self.m + 7) // 8)

    def _positions(self, key):
        data = key.encode('utf-8')
        digest = hashlib.blake2b(data, digest_size=16).digest()
        h1 = int.from_bytes(digest[:8], 'big')
        h2 = int.from_bytes(digest[8:], 'big')
        h2 = h2 % self.m
        if h2 == 0:
            h2 = 1
        for i in range(self.k):
            yield (h1 + i * h2) % self.m

    def add(self, key):
        for pos in self._positions(key):
            self.bits[pos >> 3] |= 1 << (pos & 7)

    def contains(self, key):
        for pos in self._positions(key):
            if not (self.bits[pos >> 3] & (1 << (pos & 7))):
                return False
        return True


def demo():
    n = 1000
    p = 0.01
    bf = BloomFilter(n, p)

    for i in range(n):
        bf.add(f'link:{i}')

    assert all(bf.contains(f'link:{i}') for i in range(n))

    trials = 5000
    false_positive = 0
    for i in range(trials):
        if bf.contains(f'missing:{i}'):
            false_positive += 1

    estimate = (1 - math.exp(-bf.k * n / bf.m)) ** bf.k
    print(f'm={bf.m}, k={bf.k}, 估算误判率={estimate:.4f}, 实测误判率={false_positive / trials:.4f}')


if __name__ == '__main__':
    demo()

运行方法：将代码保存为 bloom_demo.py，执行 python bloom_demo.py。示例无第三方依赖，会打印位数组大小、哈希次数、估算误判率和一次统计得到的误判率。实测值受样本影响，不应把它当成固定性能结论。

接入短链服务时，把 link:i 换成规范化后的短链 ID。创建短链成功后调用 add；请求入口先调用 contains。若返回 False，说明该 ID 一定不存在，可直接返回不存在响应；若返回 True，再查缓存和数据库。布隆过滤器只做前置拦截，不能替代缓存空值、限流和数据库权威校验。

边界与限制

• 容量：n 是预期元素数，实际插入超过 n 后误判率会上升，应留余量并监控规模。
• 删除：普通布隆过滤器不支持删除，删除短链后仍可能返回“可能存在”，需要查权威存储或定期重建；确有删除需求可考虑计数布隆过滤器。
• 误判非零：返回可能存在时，仍要查缓存和数据库；可配合空值缓存与限流降低穿透影响。
• 假阴性：正确实现下，插入过的元素不会被判为不存在；但若编码、ID 规范化或哈希算法不一致，该性质会被破坏。
• 持久化：进程重启后需要从数据库预热或持久化位图；多实例部署要共享位图或各自预热，并注意一致性。
• 并发：多线程同时写位数组时建议加锁；最小示例为单线程。
• 内存：位数组约占 m/8 字节，误判率要求越低，内存开销越大。
• 安全：不要用布隆过滤器做权限判断；哈希输入需统一大小写、空格和编码规则。