作者：EchoZenith Github

百万数据毫秒判重：一文吃透布隆过滤器的原理、实战与 Python-MySQL 落地

关键词：布隆过滤器、Bloom Filter、MySQL、Python、缓存穿透、去重、RedisBloom、MyRocks

1. 引言：为什么需要布隆过滤器？

• 缓存穿透：黑客构造千万级不存在的 key，直击数据库，拖垮服务。
• 日志去重：爬虫每天 5 亿 URL，如何毫秒级判断“是否已抓取”？
• 规则引擎：10 万条黑名单手机号，如何常驻内存且不爆表？

传统方案（HashSet、B+Tree）要么内存爆炸，要么磁盘 IO 高。
布隆过滤器（Bloom Filter）用 <1% 的内存 换取 可接受的误报率，成为“看门神”。

2. 原理解剖：一张图看懂 Bloom Filter

1. 插入：k 个哈希函数 把元素映射到 k 个位，全部置 1。
2. 查询：任意一位为 0 ⇒ 一定不存在；全 1 ⇒ 可能存在（假阳性）。

无漏报，有误报；无删除，可扩展。

3. 数学公式：如何选 m 与 k？

最优公式：

$$ m = -\frac{n \ln p}{(\ln 2)^2}, \quad k = \frac{m}{n} \ln 2 $$

快速估算表：

4. 业界实战落地路线

下文重点演示 4. 自实现。

5. Python + MySQL 手把手实战

5.1 表设计：把位图拆成 64 bit 块

CREATE TABLE bloom_bits (
    slot BIGINT UNSIGNED PRIMARY KEY,  -- 0..m/64-1
    bits BIGINT NOT NULL DEFAULT 0     -- 64 位一块
) ENGINE=InnoDB;

5.2 核心代码

import mmh3, pymysql, math

class MySQLBloomFilter:
    def __init__(self, m, k, cfg):
        self.m, self.k, self.db = m, k, pymysql.connect(**cfg, autocommit=False)
        self._init_slots()

    # 计算槽与掩码
    def _slot_bit(self, idx):
        slot, bit = divmod(idx, 64)
        return slot, 1 << bit

    # 初始化空槽
    def _init_slots(self):
        with self.db.cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM bloom_bits")
            if cur.fetchone()[0] == 0:
                slots = self.m // 64
                cur.executemany("INSERT INTO bloom_bits(slot) VALUES (%s)",
                                [(s,) for s in range(slots)])
                self.db.commit()

    # 插入
    def add(self, item: str):
        idxs = [mmh3.hash(item, seed) % self.m for seed in range(self.k)]
        slots_bits = [self._slot_bit(i) for i in idxs]
        with self.db.cursor() as cur:
            cur.execute("SELECT slot,bits FROM bloom_bits WHERE slot IN %s",
                        (tuple({s for s, _ in slots_bits}),))
            old = {s: b for s, b in cur.fetchall()}
        with self.db.cursor() as cur:
            for slot, bit in slots_bits:
                new = old.get(slot, 0) | bit
                if new != old.get(slot, 0):
                    cur.execute("UPDATE bloom_bits SET bits=%s WHERE slot=%s",
                                (new, slot))
            self.db.commit()

    # 查询
    def contains(self, item: str) -> bool:
        idxs = [mmh3.hash(item, seed) % self.m for seed in range(self.k)]
        slots_bits = [self._slot_bit(i) for i in idxs]
        with self.db.cursor() as cur:
            cur.execute("SELECT slot,bits FROM bloom_bits WHERE slot IN %s",
                        (tuple({s for s, _ in slots_bits}),))
            cache = {s: b for s, b in cur.fetchall()}
        return all((cache.get(slot, 0) & bit) != 0 for slot, bit in slots_bits)

5.3 运行示例

if __name__ == '__main__':
    cfg = dict(host='127.0.0.1', user='root', password='123456', db='test')
    n, p = 1_000_000, 0.01
    m = int(-n * math.log(p) / (math.log(2) ** 2))
    m = (m // 64 + 1) * 64
    k = int(m / n * math.log(2))
    bf = MySQLBloomFilter(m, k, cfg)

    bf.add("user123")
    print(bf.contains("user123"))  # True
    print(bf.contains("user999"))  # False

5.4 性能压测

单次 add/contains 仅 1~2 条 SQL，m=1 M 时表 < 2 MB，可全放 InnoDB Buffer。

6. 高级技巧与坑

1. 并发写热点：
   同一 64 位块被多条线程同时 UPDATE 会行锁等待 → 把 bits 拆成 更细粒度 或 异步合并。
2. 动态扩容：
   新建 bloom_bits_v2 ，双写 + 读合并，实现 滚动扩容。
3. 计数删除：
   把 BIGINT 换成 TINYINT 计数器（4 bit），实现 Counting Bloom Filter。
4. 分布式：
   多个实例共用同一张 bloom_bits ，事务隔离级别=RC 即可。
5. 哈希选择：
   MurmurHash3 速度最快；若需加密可用 SipHash。

7. 总结：一张脑图带走 Bloom Filter

布隆过滤器
├─ 原理：位数组 + k 哈希
├─ 公式：m = -n·lnp/(ln2)²
├─ 落地
│  ├─ 本地：Guava
│  ├─ 缓存：RedisBloom
│  ├─ 引擎：MyRocks
│  └─ 自研：Python+MySQL（本文）
└─ 场景：去重、防穿透、黑名单

记住口诀：“一定不存在 100% 准确，可能存在可控误报；内存降低 99%，速度进毫秒。”