弗兰的悲惨之旅
99.73M · 2026-04-04
上周线上出了事故,凌晨两点被报警电话叫醒——某个查询接口的 QPS 突然飙到平时的 20 倍,MySQL 直接扛不住了,慢查询堆满了。排查下来,是有人拿一堆不存在的 ID 疯狂请求,所有请求全部穿透 Redis 打到了数据库。
缓存穿透这个概念谁都知道,但真正线上被打一次才知道有多疼。事后把三种主流方案都实测了一遍,踩了不少坑,这篇把完整过程分享出来。
缓存穿透 = 查询的 key 在 Redis 和数据库中都不存在,每次请求都打穿到 DB。 这和缓存击穿(热 key 过期)、缓存雪崩(大量 key 同时过期)是完全不同的问题。
三种方案的适用场景直接列一下:
最终选择是布隆过滤器 + 缓存空值组合拳,下面展开说。
正常的缓存流程大家都熟:
graph LR
A[客户端请求] --> B{Redis 有缓存?}
B -->|命中| C[返回缓存数据]
B -->|未命中| D[查询 MySQL]
D --> E{数据存在?}
E -->|存在| F[写入 Redis + 返回数据]
E -->|不存在| G[返回空 / 直接返回]
G --> H[下次请求还是打到 DB]
H --> D
问题出在最后那个循环——key 不存在时不会写缓存,导致每次都穿透到 DB。如果有人故意拿 user_id = -1 或者一堆随机 UUID 来查,Redis 形同虚设,DB 直接被打穿。
我那次线上事故,攻击者用的是递增的负数 ID,简单粗暴但很有效。
最直觉的方案:查询结果为空也缓存起来,设一个较短的 TTL。
import redis
import json
r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
# 空值标记,不要用 None 或空字符串,容易和正常值混淆
EMPTY_CACHE_FLAG = "@@EMPTY@@"
EMPTY_CACHE_TTL = 120 # 空值缓存 2 分钟
NORMAL_CACHE_TTL = 3600 # 正常数据缓存 1 小时
def get_user_by_id(user_id: int) -> dict | None:
cache_key = f"user:{user_id}"
# 1. 先查 Redis
cached = r.get(cache_key)
if cached is not None:
if cached == EMPTY_CACHE_FLAG:
# 命中空值缓存,直接返回 None,不打 DB
return None
return json.loads(cached)
# 2. Redis 没有,查 DB
user = query_user_from_db(user_id)
if user is None:
# 3. DB 也没有,缓存空值
r.setex(cache_key, EMPTY_CACHE_TTL, EMPTY_CACHE_FLAG)
return None
# 4. DB 有数据,正常缓存
r.setex(cache_key, NORMAL_CACHE_TTL, json.dumps(user))
return user
def query_user_from_db(user_id: int) -> dict | None:
"""模拟数据库查询"""
# 实际项目中这里是 SQL 查询
# SELECT * FROM users WHERE id = %s
pass
用 wrk 模拟了 1000 个不存在的 ID 并发请求:
坑 1:空值标记选错了。 一开始直接缓存空字符串 "",结果有个接口正好会返回空字符串作为合法值,线上出了 bug。后来改成特殊标记字符串,在序列化层统一处理。
坑 2:TTL 设太长导致数据不一致。 如果一个用户刚注册,ID 之前被缓存了空值,那在 TTL 过期前用户查不到自己的数据。解决方案是写入数据库时主动删除对应的空值缓存:
def create_user(user_data: dict) -> int:
user_id = insert_user_to_db(user_data)
# 创建成功后,主动删除可能存在的空值缓存
r.delete(f"user:{user_id}")
return user_id
坑 3:内存爆炸。 攻击者用随机 key 来打时,每个 key 都会在 Redis 里存一条空值缓存,Redis 内存会被撑爆。这个方案对随机 key 攻击基本无效。
布隆过滤器的核心思想:用一个 bit 数组 + 多个 hash 函数,快速判断一个元素「一定不存在」或「可能存在」。
注意这个「可能存在」——布隆过滤器有误判率,会把不存在的判断为存在(false positive),但不会把存在的判断为不存在(no false negative)。这个特性刚好适合缓存穿透场景。
graph LR
A[客户端请求] --> B{布隆过滤器判断}
B -->|一定不存在| C[直接返回空, 不查 DB]
B -->|可能存在| D{Redis 有缓存?}
D -->|命中| E[返回缓存数据]
D -->|未命中| F[查询 MySQL]
F --> G[写入 Redis + 返回]
Redis 4.0+ 支持 RedisBloom 模块,用起来很方便。如果 Redis 没装这个模块,也可以用 Python 的 pybloom_live 库在应用层实现,但更推荐用 Redis 原生的,省得维护内存中的状态。
import redis
r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
BLOOM_KEY = "bf:user_ids"
def init_bloom_filter():
"""
初始化布隆过滤器
预计元素数量 100万,误判率 0.01(1%)
实际内存占用约 1.2MB,非常省
"""
try:
# BF.RESERVE key error_rate capacity
r.execute_command("BF.RESERVE", BLOOM_KEY, 0.01, 1000000)
print("布隆过滤器创建成功")
except redis.ResponseError as e:
if "item exists" in str(e):
print("布隆过滤器已存在,跳过创建")
else:
raise
def load_existing_ids():
"""把数据库中已有的 ID 全量灌入布隆过滤器"""
# 分批加载,别一次性全捞出来
batch_size = 5000
offset = 0
while True:
ids = query_user_ids_batch(offset, batch_size)
if not ids:
break
# BF.MADD 批量添加,比循环 BF.ADD 快很多
r.execute_command("BF.MADD", BLOOM_KEY, *ids)
offset += batch_size
print(f"已加载 {offset} 条 ID")
def get_user_by_id(user_id: int) -> dict | None:
cache_key = f"user:{user_id}"
# 1. 布隆过滤器前置拦截
exists = r.execute_command("BF.EXISTS", BLOOM_KEY, user_id)
if not exists:
# 布隆过滤器说不存在,那就一定不存在
return None
# 2. 可能存在,走正常缓存逻辑
cached = r.get(cache_key)
if cached is not None:
return json.loads(cached)
# 3. 查 DB
user = query_user_from_db(user_id)
if user:
r.setex(cache_key, 3600, json.dumps(user))
return user
def create_user(user_data: dict) -> int:
"""新增用户时,同步更新布隆过滤器"""
user_id = insert_user_to_db(user_data)
# 别忘了把新 ID 加入布隆过滤器!
r.execute_command("BF.ADD", BLOOM_KEY, user_id)
return user_id
同样的 1000 个不存在的 ID 并发压测:
坑 1:布隆过滤器不支持删除。 标准布隆过滤器只能添加不能删除。删了一条用户数据,布隆过滤器里还是会判断为「可能存在」,结果是多查一次 DB 返回空。删除频繁的业务可以考虑 Cuckoo Filter(CF.RESERVE/CF.ADD/CF.DEL),RedisBloom 也支持。
坑 2:全量初始化太慢。 线上 800 万用户 ID,第一次灌数据花了快 3 分钟。后来改成 pipeline 批量操作 + 后台异步加载,不阻塞主服务启动:
def load_existing_ids_with_pipeline():
"""用 pipeline 批量加载,速度提升 10 倍+"""
batch_size = 5000
offset = 0
while True:
ids = query_user_ids_batch(offset, batch_size)
if not ids:
break
pipe = r.pipeline()
for uid in ids:
pipe.execute_command("BF.ADD", BLOOM_KEY, uid)
pipe.execute()
offset += batch_size
坑 3:误判率的取舍。 误判率设 1% 意味着每 100 个不存在的 key,有 1 个会穿透到 DB。对穿透零容忍的话可以调到 0.001(0.1%),但内存会翻倍。最终选了 0.01,再配合方案一的空值缓存兜底,效果不错。
这个方案严格来说不算缓存层的解决方案,但在防恶意攻击场景下是必须做的。
import re
from functools import wraps
from collections import defaultdict
import time
# 简单的滑动窗口限流
request_counter = defaultdict(list)
def rate_limit(max_requests=100, window_seconds=60):
"""每个 IP 每分钟最多 100 次请求"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
client_ip = get_client_ip()
now = time.time()
# 清理过期记录
request_counter[client_ip] = [
t for t in request_counter[client_ip]
if now - t < window_seconds
]
if len(request_counter[client_ip]) >= max_requests:
return {"error": "rate limited"}, 429
request_counter[client_ip].append(now)
return func(*args, **kwargs)
return wrapper
return decorator
def validate_user_id(user_id) -> bool:
"""参数校验:在缓存之前就拦截明显非法的请求"""
if not isinstance(user_id, int):
return False
if user_id <= 0: # 我们的 ID 都是正整数
return False
if user_id > 10_000_000_000: # 业务上不可能有这么大的 ID
return False
return True
@rate_limit(max_requests=100, window_seconds=60)
def api_get_user(user_id):
# 1. 参数校验
if not validate_user_id(user_id):
return {"error": "invalid user_id"}, 400
# 2. 正常走缓存逻辑
user = get_user_by_id(user_id)
if user is None:
return {"error": "user not found"}, 404
return user, 200
就是在最外层把明显不合法的请求干掉,减少穿透到缓存层和 DB 层的请求量。
单一方案都有短板,线上实际是这么组合的:
graph TD
A[客户端请求] --> B[参数校验 + IP 限流]
B -->|非法请求| C[直接拒绝 400/429]
B -->|合法请求| D[布隆过滤器判断]
D -->|一定不存在| E[返回 404]
D -->|可能存在| F{Redis 缓存}
F -->|命中| G[返回数据]
F -->|未命中| H[查 MySQL]
H -->|数据存在| I[写缓存 + 返回]
H -->|数据不存在| J[缓存空值 2min + 返回 404]
三层防线:
上线后,同样的攻击流量下,DB QPS 从 3000+ 降到了个位数,Redis 内存增量控制在 2MB 以内。
| 维度 | 缓存空值 | 布隆过滤器 | 请求校验+限流 |
|---|---|---|---|
| 实现复杂度 | ⭐ 低 | ⭐⭐ 中 | ⭐ 低 |
| 内存开销 | 高(随攻击 key 增长) | 低(固定大小) | 无 |
| 对随机 key 攻击 | 基本无效 | 有效 | ️ 部分有效 |
| 数据一致性 | 需处理 TTL | 不支持删除 | 无影响 |
| 误判率 | 无 | 有(可控) | 无 |
| 推荐场景 | 穿透量小,key 范围有限 | 海量数据,key 范围大 | 所有场景(必选项) |
缓存穿透说起来简单,线上真碰到了,排查 + 修复 + 验证一套下来搞了快两天。几个关键经验:
面试被问到「缓存穿透和缓存击穿的区别」,别只背概念了——能讲出线上踩坑经历和组合方案的细节,比背八股管用多了。