布隆过滤器

解决方案概述

核心思路：在缓存层之前加一道”过滤网”，提前拦截不存在的 key

问题引入

在上一节我们学习了使用缓存空对象来防护缓存穿透。

这个方案虽然简单有效，但也存在问题：

场景：恶意攻击者构造 10 万个不存在的 key

缓存空对象方案：
- 每个不存在的 key 都缓存一个空值
- 10 万个空值占用大量内存
- 假设每个空值 100 字节，总共需要约 10MB 内存

问题：
❌ 内存浪费严重
❌ 大量无效 key 污染缓存
❌ 可能挤占正常数据的缓存空间

有没有更高效的方案，既能拦截不存在的 key，又不占用太多内存？

这就是布隆过滤器大显身手的地方。

什么是布隆过滤器？

布隆过滤器（Bloom Filter）是一种概率型数据结构，用于快速判断一个元素是否存在于集合中。

核心特点

输入：元素 x
输出：两种可能
- "一定不存在"（100% 准确）
- "可能存在"（有小幅误判率）

形象比喻：

布隆过滤器就像一个筛子：

• 能 100% 拦住明显太大的石子（一定不存在的数据）
• 但可能会让小石子漏过去（可能存在，需要进一步检查）

与其他方案对比

工作原理

布隆过滤器的结构

初始化：
- 一个长度为 m 的位数组（所有位初始为 0）
- k 个独立的哈希函数

位数组示例（m=16）：
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
 ↑                                                  ↑
位置 0                                          位置 15

添加元素

添加元素 "北京"：

1. 使用 k 个哈希函数计算位置
   哈希函数 1: hash1("北京") = 3
   哈希函数 2: hash2("北京") = 7
   哈希函数 3: hash3("北京") = 15

2. 将对应位置设为 1
[0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1]
           ↑           ↑                       ↑
          位置 3      位置 7                  位置 15

查询元素

查询 "上海"（假设上海是合法城市）：

1. 计算哈希位置
   hash1("上海") = 3
   hash2("上海") = 9
   hash3("上海") = 15

2. 检查这些位置
[0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1]
           ↑     ?                      ↑
          位置 3  位置 9（为 0）位置 15

3. 发现位置 9 为 0，说明"上海"一定不在集合中

查询 "深圳"（假设深圳是合法城市）：

1. 计算哈希位置
   hash1("深圳") = 3
   hash2("深圳") = 7
   hash3("深圳") = 15

2. 检查这些位置
[0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 1]
           ↑           ↑                       ↑
          位置 3      位置 7                  位置 15

3. 所有位置都为 1，说明"深圳"可能存在

误判的原因

假设已经添加了"北京"、"广州"、"成都"

现在查询"武汉"（实际不存在）：

hash1("武汉") = 3  ← "北京" 设置过
hash2("武汉") = 7  ← "北京" 设置过
hash3("武汉") = 15 ← "广州" 设置过

所有位置都为 1！
布隆过滤器说"可能存在"，但实际上"武汉"从未添加过

这就是误判（False Positive）

关键点：

• 布隆过滤器说”不存在” → 100% 不存在
• 布隆过滤器说”存在” → 可能误判，需要进一步检查

实战：布隆过滤器防护缓存穿透

方案一：使用 Python 布隆过滤器库

from pybloom_live import BloomFilter

# 初始化布隆过滤器
# capacity: 预计元素数量
# error_rate: 可接受的误判率（0.001 = 0.1%）
bloom = BloomFilter(capacity=10000, error_rate=0.001)

# 预先加载所有合法城市
valid_cities = ['北京', '上海', '深圳', '广州', '杭州', ...]
for city in valid_cities:
    bloom.add(city)

def get_weather(city):
    cache_key = f"weather:{city}"

    # 1. 先用布隆过滤器检查
    if city not in bloom:
        # 城市一定不存在，直接返回
        return {'error': '城市不存在', 'blocked_by': 'bloom_filter'}

    # 2. 通过检查，继续正常流程
    cached = redis.get(cache_key)
    if cached:
        return cached

    # 3. 调用外部 API
    response = external_api.get_weather(city)

    # 4. 写入缓存
    redis.setex(cache_key, 3600, response.data)
    return response.data

方案二：使用 Redis 实现布隆过滤器

Redis 4.0+ 提供了布隆过滤器模块：

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

# 初始化布隆过滤器
# name: 过滤器名称
# error_rate: 误判率（默认 0.01）
# capacity: 预计元素数量
redis_client.execute_command(
    'BF.RESERVE', 'city_filter', '0.001', '10000'
)

# 添加元素
def add_cities(cities):
    for city in cities:
        redis_client.execute_command('BF.ADD', 'city_filter', city)

# 检查元素
def city_exists(city):
    return redis_client.execute_command('BF.EXISTS', 'city_filter', city) == 1

# 使用示例
def get_weather(city):
    cache_key = f"weather:{city}"

    # 1. 检查布隆过滤器
    if not city_exists(city):
        return {'error': '城市不存在', 'blocked_by': 'bloom_filter'}

    # 2. 检查缓存
    cached = redis.get(cache_key)
    if cached:
        return cached

    # 3. 调用外部 API
    response = external_api.get_weather(city)
    redis.setex(cache_key, 3600, response.data)
    return response.data

方案三：缓存空对象 + 布隆过滤器（组合方案）

from pybloom_live import BloomFilter
import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)
NULL_MARKER = '__NULL__'

# 布隆过滤器
bloom = BloomFilter(capacity=100000, error_rate=0.001)
valid_cities = load_valid_cities()  # 加载合法城市列表
for city in valid_cities:
    bloom.add(city)

def get_weather(city):
    cache_key = f"weather:{city}"

    # 第一层防护：布隆过滤器
    if city not in bloom:
        # 城市一定不存在
        return {'error': '城市不存在', 'blocked_by': 'bloom_filter'}

    # 第二层防护：缓存空对象
    cached = redis_client.get(cache_key)

    if cached == NULL_MARKER:
        # 之前查询过，确认不存在
        return {'error': '城市不存在', 'blocked_by': 'null_cache'}

    if cached:
        # 缓存命中
        return cached

    # 第三层：调用外部 API
    try:
        response = external_api.get_weather(city)
        data = response.data
    except Exception:
        # API 调用失败，也缓存空值
        redis_client.setex(cache_key, 60, NULL_MARKER)
        raise

    if not data:
        # 数据不存在，缓存空值
        redis_client.setex(cache_key, 60, NULL_MARKER)
        return {'error': '城市不存在', 'blocked_by': 'api_check'}

    # 正常数据，缓存
    redis_client.setex(cache_key, 3600, data)
    return data

效果对比

防护效果

内存占用对比

场景：10 万个合法 key

缓存空对象方案：
- 假设攻击 10 万个不存在的 key
- 每个空值缓存 100 字节
- 总占用：10 万 × 100 字节 ≈ 10MB

布隆过滤器方案：
- 容量 10 万，误判率 0.1%
- 位数组大小：约 100 万位 ≈ 125KB
- 总占用：约 100-200KB

内存节省：约 50-100 倍！

深度思考

误判率如何设置？

# 误判率越低，需要的内存越多
BloomFilter(capacity=100000, error_rate=0.001)  # 0.1% 误判率
BloomFilter(capacity=100000, error_rate=0.0001) # 0.01% 误判率，内存翻倍

推荐设置：

• 一般场景：0.001（0.1%）
• 关键场景：0.0001（0.01%）
• 极端场景：配合缓存空对象双重防护

动态更新的挑战

布隆过滤器不支持删除元素：

bloom.add("北京")      # 添加
# bloom.remove("北京")  # ❌ 不支持！

# 如果需要删除，需要使用"计数布隆过滤器"
from pybloom_live import ScalableBloomFilter
bloom = ScalableBloomFilter(error_rate=0.001)

解决方案：

1. 使用 ScalableBloomFilter（支持动态扩展）
2. 定期重建布隆过滤器
3. 使用 Redis 布隆过滤器模块（支持 BF.DEL）

合法数据集合的维护

# 方案 1：从数据库加载
def load_valid_cities():
    cities = db.query("SELECT city_name FROM cities")
    return [c.city_name for c in cities]

# 方案 2：从配置文件加载
def load_valid_cities():
    with open('valid_cities.json') as f:
        return json.load(f)

# 方案 3：定期同步
def sync_bloom_filter():
    bloom = BloomFilter(capacity=100000, error_rate=0.001)
    cities = external_api.get_all_cities()
    for city in cities:
        bloom.add(city)
    return bloom

当前技术架构

课后练习

练习 1

布隆过滤器为什么说”不存在”是 100% 准确的？

如果元素 X 真的存在：
- 添加 X 时，它的 k 个位置都被设为 1
- 查询 X 时，这 k 个位置都为 1
- 判断：可能存在（实际确实存在）

如果元素 X 不存在：
- X 的 k 个哈希位置中，至少有一个从未被设置
- 查询 X 时，至少有一个位置为 0
- 判断：一定不存在（100% 准确）

练习 2

布隆过滤器的误判率如何影响内存占用？

误判率越低 → 需要的位数组越长 → 内存占用越多

公式：
m = - (n × ln p) / (ln 2)²

其中：
- m: 位数组长度（比特）
- n: 预计元素数量
- p: 误判率
- ln: 自然对数

练习 3

请实现一个使用布隆过滤器防护缓存穿透的完整代码。

from pybloom_live import BloomFilter
import redis
import json

# 初始化
redis_client = redis.Redis(host='localhost', port=6379, db=0)
NULL_MARKER = '__NULL__'

# 布隆过滤器（容量 10 万，误判率 0.1%）
bloom = BloomFilter(capacity=100000, error_rate=0.001)

# 加载合法城市列表
def init_bloom_filter():
    valid_cities = external_api.get_all_cities()
    for city in valid_cities:
        bloom.add(city)

# 带布隆过滤器的天气查询
def get_weather(city):
    cache_key = f"weather:{city}"

    # 1. 布隆过滤器检查（第一层防护）
    if city not in bloom:
        # 城市一定不存在
        return {
            'error': '城市不存在',
            'blocked_by': 'bloom_filter'
        }

    # 2. 检查缓存空值标记（第二层防护）
    cached = redis_client.get(cache_key)

    if cached == NULL_MARKER:
        return {
            'error': '城市不存在',
            'blocked_by': 'null_cache'
        }

    # 3. 缓存命中
    if cached:
        return json.loads(cached)

    # 4. 调用外部 API
    try:
        response = external_api.get_weather(city)
        data = response.data
    except Exception:
        # API 调用失败，缓存空值
        redis_client.setex(cache_key, 60, NULL_MARKER)
        raise

    # 5. 写入缓存
    if not data:
        redis_client.setex(cache_key, 60, NULL_MARKER)
        return {'error': '城市不存在'}

    redis_client.setex(cache_key, 3600, json.dumps(data))
    return data

# 初始化
init_bloom_filter()