布隆过滤器

1. 布隆过滤器原理

1.1 基本结构

布隆过滤器是一个空间高效的概率数据结构，用于判断元素是否可能存在：

位数组 m 位 + k 个独立哈希函数

添加元素 x:
  h1(x) % m → 位置设为1
  h2(x) % m → 位置设为1
  ...
  hk(x) % m → 位置设为1

查询元素 y:
  检查 h1(y), h2(y), ..., hk(y) 对应位置是否全为1
  全为1 → 可能存在（有误判率）
  有0   → 一定不存在

1.2 特性

2. 误判率分析

2.1 数学推导

设位数组大小 $m$，哈希函数个数 $k$，已插入元素数 $n$：

某个位仍为 0 的概率：

$P(\text{位为0}) = \left(1 - \frac{1}{m}\right)^{kn} \approx e^{-kn/m}$

误判率（所有 $k$ 个位都为 1）：

$P(\text{误判}) = \left(1 - e^{-kn/m}\right)^k$

2.2 最优哈希函数个数

对误判率求导，令其为 0：

$k_{\text{opt}} = \frac{m}{n} \ln 2$

2.3 最优位数组大小

给定目标误判率 $\epsilon$：

$m = -\frac{n \ln \epsilon}{(\ln 2)^2}$

2.4 常用参数表

3. 实现

3.1 Python 实现

import mmh3
import math

class BloomFilter:
    def __init__(self, capacity, error_rate=0.01):
        self.capacity = capacity
        self.error_rate = error_rate
        self.bit_size = self._optimal_bit_size(capacity, error_rate)
        self.hash_count = self._optimal_hash_count(self.bit_size, capacity)
        self.bit_array = bytearray(math.ceil(self.bit_size / 8))

    @staticmethod
    def _optimal_bit_size(n, p):
        return int(-n * math.log(p) / (math.log(2) ** 2))

    @staticmethod
    def _optimal_hash_count(m, n):
        return int(m / n * math.log(2))

    def _set_bit(self, index):
        byte_index = index // 8
        bit_offset = index % 8
        self.bit_array[byte_index] |= (1 << bit_offset)

    def _get_bit(self, index):
        byte_index = index // 8
        bit_offset = index % 8
        return bool(self.bit_array[byte_index] & (1 << bit_offset))

    def add(self, item):
        for seed in range(self.hash_count):
            index = mmh3.hash(str(item), seed) % self.bit_size
            self._set_bit(index)

    def contains(self, item):
        for seed in range(self.hash_count):
            index = mmh3.hash(str(item), seed) % self.bit_size
            if not self._get_bit(index):
                return False
        return True  # 可能误判

3.2 Counting Bloom Filter（支持删除）

class CountingBloomFilter:
    def __init__(self, capacity, error_rate=0.01):
        m = int(-capacity * math.log(error_rate) / (math.log(2) ** 2))
        k = int(m / capacity * math.log(2))
        self.counters = [0] * m
        self.hash_count = k
        self.bit_size = m

    def add(self, item):
        for seed in range(self.hash_count):
            idx = mmh3.hash(str(item), seed) % self.bit_size
            self.counters[idx] += 1

    def remove(self, item):
        if not self.contains(item):
            return False
        for seed in range(self.hash_count):
            idx = mmh3.hash(str(item), seed) % self.bit_size
            self.counters[idx] -= 1
        return True

    def contains(self, item):
        for seed in range(self.hash_count):
            idx = mmh3.hash(str(item), seed) % self.bit_size
            if self.counters[idx] == 0:
                return False
        return True

4. 工程应用

4.1 缓存穿透防护

# 预加载所有有效ID到布隆过滤器
bf = BloomFilter(capacity=10_000_000, error_rate=0.001)
for user_id in db.query("SELECT id FROM users"):
    bf.add(user_id)

def get_user(user_id):
    if not bf.contains(user_id):
        return None  # 一定不存在，直接返回
    # 查缓存 → 查数据库

4.2 爬虫 URL 去重

visited = BloomFilter(capacity=100_000_000)

def crawl(url):
    if visited.contains(url):
        return  # 可能已访问
    visited.add(url)
    # 爬取页面...

4.3 Redis 布隆过滤器模块

# Redis 4.0+ BF 模块
BF.RESERVE my_filter 0.001 1000000
BF.ADD my_filter "user:1001"
BF.EXISTS my_filter "user:1001"  → 1
BF.EXISTS my_filter "user:9999"  → 0