布隆过滤器


布隆过滤器

布隆过滤器是一种空间效率极高、时间复杂度低的概率型数据结构,用于快速判断 “元素是否在集合中”,特点是无假阴性(不在则一定返回不在)、可能有假阳性(在则可能误判),无法直接删除元素。

1. 核心定义

布隆过滤器由一个二进制位数组(初始全为 0)和多个独立的哈希函数组成。它不存储元素本身,仅通过哈希映射标记元素的存在状态,以此实现高效的查询和插入。


2. 工作原理

  1. 初始化:创建长度为 m 的二进制位数组,同时确定 k 个不同的哈希函数(哈希函数个数需根据元素量和误判率估算)。
  2. 插入元素:将元素传入 k 个哈希函数,得到 k 个不同的哈希值,再将位数组中对应哈希值的下标位置设为 1。
  3. 查询元素:同样用 k 个哈希函数处理待查询元素,若所有对应下标的位都为 1,则元素 “可能在集合中”;若有任意一位为 0,则元素 “一定不在集合中”。

3. 核心优缺点

优点

  • 空间效率极高:仅用二进制位存储状态,无需存储元素本身,比哈希表、红黑树等节省几个数量级的空间。
  • 时间复杂度低:插入和查询的时间复杂度均为 O (k)(k 为哈希函数个数,通常是常数),效率接近 O (1)。

缺点

  • 存在假阳性:无法 100% 准确判断 “元素在集合中”,只能给出 “可能在” 的结论(误判率可通过参数调整降低)。
  • 普通版无法删除:位数组的位是共享的,删除一个元素会将对应位设为 0,可能影响其他元素的判断。
  • 需提前预估参数:需先确定预期元素数量 n 和可接受误判率 p,才能计算出合适的位数组长度 m 和哈希函数个数 k。

4. 实现代码

实现中使用了 mmh3 哈希库(非加密哈希,速度快)和 bitarray 库(高效存储二进制位),需先安装依赖。

import math
import mmh3
from bitarray import bitarray


class BloomFilter:
    def __init__(self, expected_num: int, false_positive_rate: float):
        """
        初始化布隆过滤器
        :param expected_num: 预期插入的元素数量(n)
        :param false_positive_rate: 可接受的误判率(p),如0.01表示1%
        """
        self.expected_num = expected_num  # 预期元素数n
        self.false_positive_rate = false_positive_rate  # 误判率p

        # 计算位数组长度m:m = - (n * ln(p)) / (ln(2)^2)
        self.bit_size = int(- (expected_num * math.log(false_positive_rate)) / (math.log(2) ** 2))
        # 计算哈希函数个数k:k = (m / n) * ln(2)
        self.hash_func_num = int((self.bit_size / expected_num) * math.log(2))

        # 初始化位数组(全为0)
        self.bit_array = bitarray(self.bit_size)
        self.bit_array.setall(0)

        # 记录实际插入的元素数量(用于后续可能的动态调整,可选)
        self.actual_num = 0

    def add(self, item) -> None:
        """插入元素到布隆过滤器"""
        if self.actual_num >= self.expected_num:
            print(f"警告:实际元素数已超过预期值{n},误判率可能升高!")

        # 对元素进行k次哈希,得到k个索引
        for seed in range(self.hash_func_num):
            # 用mmh3哈希(支持种子,不同种子=不同哈希函数),取绝对值后对bit_size取模
            hash_value = mmh3.hash(str(item), seed)
            index = abs(hash_value) % self.bit_size
            self.bit_array[index] = 1  # 标记为1

        self.actual_num += 1

    def contains(self, item) -> bool:
        """查询元素是否可能存在(True=可能存在,False=一定不存在)"""
        for seed in range(self.hash_func_num):
            hash_value = mmh3.hash(str(item), seed)
            index = abs(hash_value) % self.bit_size
            if self.bit_array[index] == 0:
                return False  # 有一位为0,一定不存在
        return True  # 所有位为1,可能存在


# 测试代码
if __name__ == "__main__":
    # 初始化:预期插入1000个元素,误判率控制在1%
    bloom = BloomFilter(expected_num=1000, false_positive_rate=0.01)

    # 插入1000个元素
    for i in range(1000):
        bloom.add(i)

    # 测试已插入的元素(应全部返回True)
    error = 0
    for i in range(1000):
        if not bloom.contains(i):
            error += 1
    print(f"已插入元素的误判率(假阴性):{error/1000:.2%}")  # 理论上应为0%

    # 测试未插入的元素(可能存在假阳性)
    false_positive = 0
    test_num = 10000  # 测试10000个未插入的元素
    for i in range(1000, 1000 + test_num):
        if bloom.contains(i):
            false_positive += 1
    print(f"未插入元素的实际误判率:{false_positive/test_num:.2%}")  # 应接近1%

5. 典型应用场景

  • 缓存穿透防护:数据库查询前用布隆过滤器判断 key 是否存在,过滤不存在的 key,避免穿透到数据库。
  • 海量数据去重:如爬虫 URL 去重、邮件黑名单过滤、分布式系统中重复请求拦截。
  • 数据库优化:如 HBase、Redis 等数据库用它快速判断数据是否在表 / 桶中,减少无效 IO。
  • 推荐系统去重:过滤用户已浏览过的内容,避免重复推荐。

0 条评论

发表评论

暂无评论,欢迎发表您的观点!