Python中如何实现Bloomfilter？小白求指点

提个小建议，README 里面，ops 的 ps 已经有每秒的意思，就不需要 /s 了。

h691938207 2楼

在Python里实现一个简单的Bloom Filter不难，核心就三部分：一个大的比特数组（位向量）、几个哈希函数，以及“添加”和“查询”两个方法。下面我直接给你一个能跑的代码，你一看就懂。

import mmap
import hashlib
from bitarray import bitarray
from math import log

class BloomFilter:
    def __init__(self, capacity, error_rate=0.001):
        """
        初始化布隆过滤器
        :param capacity: 预期要存放的元素数量
        :param error_rate: 可接受的误判率
        """
        # 计算需要的比特数组大小 (m) 和哈希函数个数 (k)
        self.capacity = capacity
        self.error_rate = error_rate
        self.num_bits = int(-(capacity * log(error_rate)) / (log(2) ** 2))  # 公式计算位数组大小
        self.num_hashes = int((self.num_bits / capacity) * log(2))  # 公式计算哈希函数个数

        # 初始化位数组，所有位设为0
        self.bit_array = bitarray(self.num_bits)
        self.bit_array.setall(0)

    def _hash(self, item, seed):
        """使用双重哈希（或多次哈希）模拟多个独立的哈希函数"""
        # 先用一个哈希函数（如MD5）生成一个大的摘要
        hash_obj = hashlib.md5(item.encode())
        digest = int(hash_obj.hexdigest(), 16)
        # 然后通过不同的种子（seed）来生成不同的哈希值
        hash_val = (digest + seed) % self.num_bits
        return hash_val

    def add(self, item):
        """将一个元素添加到布隆过滤器中"""
        for i in range(self.num_hashes):
            # 对每个哈希函数，计算在位数组中的位置，并设为1
            index = self._hash(item, i)
            self.bit_array[index] = 1

    def __contains__(self, item):
        """检查一个元素是否可能在布隆过滤器中（可能有误判）"""
        for i in range(self.num_hashes):
            index = self._hash(item, i)
            if not self.bit_array[index]:
                # 只要有一个位置是0，肯定不存在
                return False
        # 所有位置都是1，可能存在（但有误判可能）
        return True

# ---------- 使用示例 ----------
if __name__ == "__main__":
    # 创建一个预期存放10000个元素，误判率0.1%的布隆过滤器
    bf = BloomFilter(capacity=10000, error_rate=0.001)

    # 添加一些元素
    items_to_add = ["apple", "banana", "cherry", "date"]
    for item in items_to_add:
        bf.add(item)
        print(f"已添加: {item}")

    # 测试查询
    test_items = ["apple", "banana", "grape", "fig"]
    for item in test_items:
        if item in bf:
            print(f"'{item}' 可能在集合中（或发生了误判）")
        else:
            print(f"'{item}' 肯定不在集合中")

代码解释：

初始化 (__init__): 根据你预期的元素数量 (capacity) 和可接受的误判率 (error_rate)，用公式计算出需要多大的比特数组 (num_bits) 和多少个哈希函数 (num_hashes)。这里用了 bitarray 库来高效地管理大的位数组，你需要先 pip install bitarray。
哈希函数 (_hash): 这是关键。我们不可能真的写N个不同的哈希算法。这里用一个“种子”技巧：先用MD5（或SHA256）生成一个大的整数摘要，然后通过加不同的种子 (seed) 并取模，来模拟多个独立的哈希函数的结果。num_hashes 是多少，我们就循环多少次，每次种子 i 不同。
添加 (add): 添加一个元素时，用所有哈希函数算出N个位置，把位数组中这些位置都设为1。
查询 (__contains__): 查询时，同样用所有哈希函数算出N个位置。如果任何一个位置是0，那这个元素肯定没添加过。如果所有位置都是1，那这个元素可能添加过（但也可能是其他元素把这些位碰巧都设成了1，这就是“误判”）。