4、布隆过滤器：原理、误判率与Python实现

布隆过滤器这东西，我第一次接触是在处理亿级用户ID去重的时候。当时内存快爆了，常规的Set根本扛不住。嗯，布隆过滤器就是来解决这类问题的——用极小的内存代价，判断一个元素「一定不存在」或者「可能存在」。

4.1 布隆过滤器的核心原理

说白了，布隆过滤器就是一个位数组加上多个哈希函数。它不存数据本身，只存数据的「指纹」。

我画了一张图，帮你理解这个过程：

4.2 误判率是怎么来的？

你想想看，多个元素插入后，位数组上很多位置都变成了1。查询一个没插入过的元素时，它对应的几个哈希位置可能恰好都被其他元素置1了——这就产生了假阳性。

误判率公式长这样：

p = (1 - e^(-k * n / m))^k

其中：
m = 位数组长度（bit数）
n = 插入元素数量
k = 哈希函数个数

我在项目中踩过一个坑：哈希函数数量不是越多越好。太多哈希函数会让位数组迅速被填满，误判率反而上升。

4.3 误判率计算实战

假设我们要处理1亿个元素，允许1%的误判率，需要多少内存？

误判率 p	位数组大小 m	哈希函数数 k	内存占用
10% (0.1)	4.8 × n	3	约57 MB
1% (0.01)	9.6 × n	7	约114 MB
0.1% (0.001)	14.4 × n	10	约172 MB
0.01% (0.0001)	19.2 × n	13	约229 MB

看到没？1亿条数据，用1%误判率，只要114MB。换成HashSet的话，至少得几个GB。这就是布隆过滤器的魅力。

4.4 Python实现布隆过滤器

我个人习惯用 bitarray 库来实现，性能好、内存省。下面是一个完整的实现：

import math
import hashlib
import bitarray

class BloomFilter:
    def __init__(self, n, p=0.01):
        """
        n: 预期插入元素数量
        p: 期望误判率（默认1%）
        """
        # 计算最优位数组大小
        self.m = int(-n * math.log(p) / (math.log(2) ** 2))
        # 计算最优哈希函数数量
        self.k = int(self.m / n * math.log(2))

        self.bit_array = bitarray.bitarray(self.m)
        self.bit_array.setall(0)
        self.n = n
        self.p = p

    def _hash(self, item, seed):
        """使用不同的seed生成多个哈希值"""
        hash_str = f"{seed}:{item}"
        return int(hashlib.md5(hash_str.encode()).hexdigest(), 16) % self.m

    def add(self, item):
        """插入元素"""
        for i in range(self.k):
            pos = self._hash(item, i)
            self.bit_array[pos] = 1

    def contains(self, item):
        """检查元素是否存在"""
        for i in range(self.k):
            pos = self._hash(item, i)
            if self.bit_array[pos] == 0:
                return False
        return True

    def get_false_positive_rate(self):
        """计算当前理论误判率"""
        return (1 - math.exp(-self.k * self.n / self.m)) ** self.k

# 使用示例
bf = BloomFilter(n=1000000, p=0.01)
print(f"位数组大小: {bf.m} bits ({bf.m/8/1024/1024:.2f} MB)")
print(f"哈希函数数: {bf.k}")
print(f"理论误判率: {bf.get_false_positive_rate():.4%}")

# 插入100万条数据
for i in range(1000000):
    bf.add(f"user_{i}")

# 测试误判率
false_positives = 0
test_count = 100000
for i in range(1000000, 1000000 + test_count):
    if bf.contains(f"user_{i}"):
        false_positives += 1

print(f"实际误判率: {false_positives/test_count:.4%}")

4.5 避坑指南

我曾经在一个实时去重系统里用了布隆过滤器，上线后发现误判率飙升到15%。排查了半天，原来是哈希函数写得有问题——多个seed生成的哈希值高度相关，导致位数组分布不均匀。

几个关键点：

• 哈希函数要独立：用不同的seed对同一个字符串做哈希，保证结果均匀分布
• 位数组大小要预留余量：实际插入量可能超过预期，建议预留20%~30%的容量
• 误判率不是摆设：如果业务上不能容忍任何误判，布隆过滤器只能做前置过滤，后面还得跟精确去重
• 序列化要考虑：位数组和参数(m, k, n)要一起保存，否则恢复后无法正确工作