Ru果你曾在凌晨被监控报警拉醒，kan到 MySQL 的 QPS 像坐了火箭一样飙升，那么「缓存穿透」一定是罪魁祸首。本文把概念、常见套路、代码实现以及踩坑经验全部搬上桌面帮助你在Zui短时间内把这颗定时炸弹拆掉。

Zui经典的流程图大致如下：

graph LR
    A --> B{Redis 中有对应键吗？}
    B -->|命中| C
    B -->|未命中| D
    D --> E{查询结果是否存在？}
    E -->|存在| F
    E -->|不存在| G
    G --> H
    H --> D

当请求的 key 本身根本不存在上图右侧的「G」分支会一直循环——每一次dou要去磁盘数据库打一次 SQL，久而久之 DB 承受不住。

空值缓存 + 短 TTL把查询不到的结果也放进 Redis，只是设个几分钟的过期时间。

布隆过滤器预判利用位数组快速判断「一定不存在」或「可Neng存在」，把必然为空的请求直接拦下来。

限流+参数校验在业务入口层面先剔除明显非法或频率过高的请求。

单靠其中一种往往难以满足「高并发+恶意攻击」双重需求，下面我们把它们组合成一个「三层防线」方案。

布隆过滤器本质是一个超大 bit 位图 + 多个哈希函数，Neng够在 O 时间内给出「一定不存在」或「可Neng存在」的答案。误判率可调，但不会出现「一定存在却被误判为不存在」的情况，这恰好匹配我们的需求。

import redis
import json
r = redis.Redis
BF_KEY = "bf:user_ids"
def init_bloom:
    """创建容量为 100 万、误判率 0.01% 的过滤器"""
    try:
        r.execute_command
        print
    except redis.ResponseError as e:
        if "item exists" in str:
            print
        else:
            raise
def preload_ids:
    """分批读取Yi有用户 ID 并批量写入"""
    batch = 5000
    offset = 0
    while True:
        ids = query_user_ids   # ← 自己实现的 DB 分页查询
        if not ids:
            break
        r.execute_command
        offset += batch
        print

业务查询时先走过滤器：

EMPTY_FLAG = "@@EMPTY@@"
EMPTY_TTL   = 120          # 秒，针对空值
NORMAL_TTL  = 3600         # 秒，针对正常数据
def get_user:
    # 第一步：过滤器快速排除肯定不存在的 key
    if r.execute_command == 0:
        return None
    cache_key = f"user:{uid}"
    raw = r.get
    # 第二步：命中缓存
    if raw is not None:
        return None if raw == EMPTY_FLAG else json.loads
    # 第三步：真正落库查询
    user = query_user_from_mysql
    if user is None:
        r.setex   # 写入短暂空值防止瞬时击穿
        return None
    r.setex)
     return user

即便布隆过滤器Yi经hen强大，仍然建议在 API Zui外层加一层粗粒度检查：

参数合法性：只接受正整数且不超过业务上可Neng出现的Zui大 ID。

IP/用户级别滑动窗口限流：防止同一来源短时间内发起大量请求。

from collections import defaultdict
import time
REQ_RECORD = defaultdict
def rate_limit:
    def decorator:
        def wrapper:
            ip = get_client_ip
            now = time.time
            # 清理Yi过期记录
            REQ_RECORD =  if now - t = max_req:
                return {"error": "请求太频繁，请稍后再试"}, 429
            REQ_RECORD.append
            return func
        return wrapper
    return decorator
@rate_limit
def api_get_user:
     if not isinstance or uid <= 0 or uid> 10_000_000_000:
         return {"error": "非法参数"}, 400
     data = get_user
     if data is None:
         return {"error": "用户未找到"}, 404
     return data, 200

下面两组实验分别演示了「未加任何防护」和「全链路防御」时同一批 10 万个不存在的 user_id 在 wrk 中并发模拟得到的数据：

Ke以kan到，在开启三层防线后数据库压力骤降至个位数，而 Redis 的额外占用仅几百 KB——这正是我们追求的“低成本高收益”。

误判率为 %0.001% 时每万条无效 key 大约会产生一次误报；Ru果业务对每一次 DB 请求dou极其敏感，Ke以调到 %0.0001%，但内存会随之翻倍。实际项目里我把阈值设为 %0.01% 并配合空值缓存，以免出现“误报导致 DB 被敲”的尴尬局面。

Cuckoo Filter（Cf.ADD / CF.DEL) 支持删除。Ru果业务涉及频繁软删/硬删，可考虑使用 RedisBloom 提供的 Cuckoo 模块；否则只需要在删除操作后让 DB 返回空值即可，让下一次查询走空值缓存路径。

I/O 层面经常有人用空字符串 ` 或者 Python 的 ``None`` Zuo占位，这会和真实业务返回冲突。推荐使用独一无二且不易出现的串，例如上文中的 "@@EMPTY@@"`，并统一封装序列化/反序列化函数来处理。

def create_user:
     uid = insert_user
     r.delete   # 移除可Neng残留的空标记
     r.execute_command
     return uid

def bulk_load:
     batch=10_000; offset=0
     while True:
         ids=query_user_ids
         if not ids: break
         pipe=r.pipeline
         pipe.execute_command
         pipe.execute
         offset+=batch
         print