第19章：FastAPI集成Redis：缓存与热点数据治理

1. 项目背景

业务场景

"电商秒杀"开始了。运营推送了一条消息——"机械键盘限时 5 折！"瞬间 5000 人涌入商品详情页。

小钱的监控屏幕瞬间飘红：

• 数据库 CPU 从 15% 飙到 95%
• 商品查询接口 P95 延迟从 50ms 飙到 8000ms
• PostgreSQL 连接池耗尽，新请求排队等连接
• 同一个商品详情页，每秒被查询 300 次——但商品信息根本没变过

小钱紧急加了几台数据库只读副本，但问题依旧——因为 5000 人看的是同一个商品，每次请求都要查数据库。这就是典型的热点数据问题。

更糟的是，Redis 缓存虽然被引入，但上线后发现：

• 缓存虽然有，但缓存命中率只有 12%——因为 TTL 设置太短（10 秒），缓存刚写入就过期了。
• 某天凌晨，缓存全部过期，5000 个请求同时穿透缓存打到数据库——数据库直接 OOM 宕机（缓存雪崩）。
• 有人用脚本遍历商品 ID（product_id=1,2,3…），这些 ID 在数据库中不存在——但每次查询都穿透缓存，数据库全表扫描。这就是缓存穿透攻击。
• Redis 连接池默认 max_connections=10，高并发下爆了。

痛点

粗暴地加一个 Redis 不难，难的是用对缓存。缓存的三个经典灾难：

2. 项目设计

场景：秒杀结束后的事故复盘会。小钱的数据库被打挂了 3 次。大师在白板上画了三个攻击模式。

小胖：（抱着薯片）“我就加了 3 行代码把商品信息放 Redis，怎么还能把数据库打挂？缓存不就是挡在数据库前面的一堵墙吗？”

小白：“墙也有漏洞。你把 Redis 看成一面砖墙——缓存穿透是有人在墙上钻了个洞（查不存在的 key），缓存击穿是墙有一块砖碎了（热点 key 过期），缓存雪崩是整面墙倒了（大量 key 同时过期）。”

大师：（在墙上画了三个大洞）“小白这个比喻绝了。我们今天就是来做’补墙’的——三个方案：”

小胖：“互斥锁？就是说——热点 key 过期时，只让一个请求去查数据库重建缓存，其他请求先等着？”

大师：“对。本质上是用 Redis 的 SETNX（SET if Not eXists）实现分布式锁：”

async def get_product_with_lock(product_id: int) –> dict:
cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"

# 1. 查缓存
data = await redis.get(cache_key)
if data:
return json.loads(data)

# 2. 缓存未命中 → 尝试获取互斥锁
if await redis.set(lock_key, "1", nx=True, ex=5): # SETNX + 5s过期
try:
# 3. 获取锁成功 → 查数据库（只有一个人查！）
product = await db.execute(select(Product).where(Product.id == product_id))
data = product.scalar_one_or_none()
if data:
# 4. 写缓存（随机 TTL 避免雪崩）
ttl = 3600 + random.randint(0, 600)
await redis.set(cache_key, json.dumps(data), ex=ttl)
return data
finally:
await redis.delete(lock_key) # 释放锁
else:
# 5. 没抢到锁 → 等待 100ms 后再查缓存
await asyncio.sleep(0.1)
return await get_product_with_lock(product_id) # 递归重试

技术映射：互斥锁（Mutex Lock）解决了缓存击穿——同一时刻只有一个请求重建缓存。这被称为"缓存回源保护"。Redis 的 SET key value NX EX seconds 原子操作确保锁的获取是互斥的。

小白：“那缓存穿透呢？如果有人用脚本遍历 product_id 从 1 到 100000，里面 90% 的 ID 都不存在——怎么防？”

大师："两个方案组合：

小胖：“布隆过滤器听起来很高级——它怎么实现的？”

大师：“原理很简单——用多个哈希函数把一个 key 映射到一组 bit 位上。初始化时把所有存在的 product_id 都映射进去。查询时如果任何一个 bit 为 0，说明这个 key 一定不存在。如果全为 1，说明可能存在（有一定的误判率，但绝不会漏判）。Redis 4.0+ 内置了 BF.ADD 和 BF.EXISTS 命令。”

小白：“那如果布隆过滤器说可能存在，但数据库查了也没有——这不还是穿透了吗？”

大师：“对，这就是布隆过滤器的误判率问题。通过调整参数（bit 数组大小 × 哈希函数数量），可以把误判率控制在 0.1% 以下。那 0.1% 的穿透请求——还有空值缓存兜底。”

3. 项目实战——为商品详情接口构建三级缓存防护

环境准备

pip install redis==5.2.0 # redis-py 5.x 内置 asyncio 支持
# 启动 Redis
docker run -d –name redis-test -p 6379:6379 redis:7-alpine

分步实现

步骤一：创建异步 Redis 客户端（目标：连接池 + 优雅关闭）

app/infrastructure/cache.py：

import json
from typing import Any
import redis.asyncio as aioredis
from app.core.config import settings

class RedisCache:
"""异步 Redis 缓存客户端"""

def __init__(self):
self._redis: aioredis.Redis | None = None

async def connect(self):
if self._redis is None:
self._redis = await aioredis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
max_connections=50, # 连接池上限
socket_keepalive=True,
socket_connect_timeout=5,
retry_on_timeout=True,
health_check_interval=30, # 每 30s 检查连接健康
)

async def disconnect(self):
if self._redis:
await self._redis.aclose()
self._redis = None

async def get(self, key: str) –> Any:
value = await self._redis.get(key)
return json.loads(value) if value else None

async def set(self, key: str, value: Any, ttl: int = 3600):
await self._redis.set(key, json.dumps(value, default=str), ex=ttl)

async def delete(self, key: str):
await self._redis.delete(key)

async def setnx(self, key: str, value: str, ttl: int = 5) –> bool:
"""SET if Not eXists — 互斥锁的基础操作"""
return bool(await self._redis.set(key, value, nx=True, ex=ttl))

async def delete_pattern(self, pattern: str):
"""删除匹配 pattern 的所有 key（慎用！大 key 集合会阻塞）"""
keys = await self._redis.keys(pattern)
if keys:
await self._redis.delete(*keys)

# 全局缓存实例
cache = RedisCache()

步骤二：在应用生命周期中管理 Redis 连接

app/bootstrap.py（加分）：

from contextlib import asynccontextmanager
from app.infrastructure.cache import cache

@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时：连接 Redis
await cache.connect()
yield
# 关闭时：断开 Redis
await cache.disconnect()

app = FastAPI(lifespan=lifespan, ...)

步骤三：实现带互斥锁和随机 TTL 的缓存商品查询

app/domains/product/service.py（核心）：

import asyncio
import random
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.product import Product
from app.infrastructure.cache import cache

class ProductService:
"""带缓存防护的商品服务"""

NULL_TTL = 60 # 空值缓存 TTL（防穿透）
BASE_TTL = 3600 # 基础 TTL（防击穿）
RANDOM_TTL = 600 # 随机增量 TTL（防雪崩）

async def get_product(self, db: AsyncSession, product_id: int) –> dict | None:
"""三级缓存防护的商品查询"""

cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"

# ═══ 第一层：查缓存 ═══
data = await cache.get(cache_key)
if data is not None:
if data == "__NULL__":
return None # 空值缓存命中——确认商品不存在
return data

# ═══ 第二层：互斥锁重建缓存 ═══
got_lock = await cache.setnx(lock_key, "1", ttl=10)
if got_lock:
try:
# 双重检查：获取锁后再次查缓存（可能别的请求已经重建了）
data = await cache.get(cache_key)
if data is not None:
return None if data == "__NULL__" else data

# 查数据库
product = (
await db.execute(select(Product).where(Product.id == product_id))
).scalar_one_or_none()

if product:
product_dict = {
"id": product.id,
"name": product.name,
"price": float(product.price),
"stock": product.stock,
}
ttl = self.BASE_TTL + random.randint(0, self.RANDOM_TTL)
await cache.set(cache_key, product_dict, ttl=ttl)
return product_dict
else:
# 空值缓存——防止穿透
await cache.set(cache_key, "__NULL__", ttl=self.NULL_TTL)
return None

finally:
await cache.delete(lock_key) # 释放锁

# ═══ 第三层：没抢到锁 → 等待后重试 ═══
await asyncio.sleep(0.05)
# 递归重试（最多 3 次）
return await self._retry_get(db, product_id, retries=2)

async def _retry_get(self, db: AsyncSession, product_id: int, retries: int):
try:
cache_key = f"product:{product_id}"
data = await cache.get(cache_key)
if data is not None:
return None if data == "__NULL__" else data
if retries > 0:
await asyncio.sleep(0.1)
return await self._retry_get(db, product_id, retries – 1)
return None
except Exception:
return None

async def invalidate_cache(self, product_id: int):
"""当商品信息更新时，主动删除缓存"""
await cache.delete(f"product:{product_id}")

步骤四：更新商品时主动失效缓存（Cache Aside 模式）

class ProductService:
# … 上面代码 …

async def update_product(self, db: AsyncSession, product_id: int, update_data: dict):
"""更新商品——遵循 Cache Aside：先更新数据库，再删除缓存"""
product = (await db.execute(
select(Product).where(Product.id == product_id)
)).scalar_one_or_none()

if not product:
raise NotFoundException(f"商品 {product_id} 不存在")

for key, value in update_data.items():
if value is not None:
setattr(product, key, value)
await db.commit()

# 删除缓存（而非更新缓存）——下一次读取时自动重建
await self.invalidate_cache(product_id)

return product

为什么是"删缓存"而非"改缓存"？ 读操作远多于写操作。更新时新写缓存——但可能永远不会被读。淘汰缓存（lazy delete）——下次读时按需重建，避免写无用的缓存。

步骤五：验证缓存穿透、击穿、雪崩的防护

# 启动服务
uvicorn app.main:app –reload

# 1. 正常查询——第一次穿透到 DB，第二次命中缓存
time curl -s http://localhost:8000/api/v1/products/1
# 第一次: ~50ms（查 DB）
time curl -s http://localhost:8000/api/v1/products/1
# 第二次: ~5ms（命中 Redis）

# 2. 缓存穿透验证——不存在的商品 ID，空值缓存保护
time curl -s http://localhost:8000/api/v1/products/99999
# 第一次: ~30ms（查 DB，发现不存在，写空值缓存）
time curl -s http://localhost:8000/api/v1/products/99999
# 第二次: ~5ms（命中空值缓存，直接返回 404）

# 3. 并发击穿验证——100 并发请求同一个过期 key
# 用 hey 或 ab 工具压测：
# hey -n 100 -c 100 http://localhost:8000/api/v1/products/1
# 观察数据库日志：只应有一次 SELECT 查询（互斥锁生效）

# 4. 缓存雪崩验证——更新商品时主动失效缓存
curl -s -X PATCH http://localhost:8000/api/v1/products/1 \\
-H "Authorization: Bearer $TOKEN" \\
-H "Content-Type: application/json" \\
-d '{"price": 199.00}'
# 更新成功后，缓存被删除；下次查询自动重建，TTL 随机

完整代码清单

本章完整代码见 column/code/chapter19/，主要文件：

• app/infrastructure/cache.py：异步 Redis 客户端 + 连接池
• app/domains/product/service.py：三级缓存防护的商品服务
• app/bootstrap.py：Lifespan 管理 Redis 连接

测试验证

import pytest
from unittest.mock import AsyncMock, patch
from app.domains.product.service import ProductService

@pytest.mark.asyncio
async def test_cache_hit_avoids_db():
service = ProductService()
db = AsyncMock()

with patch.object(cache, 'get', new_callable=AsyncMock) as mock_get:
mock_get.return_value = {"id": 1, "name": "Keyboard", "price": 99.0}

result = await service.get_product(db, 1)

assert result["name"] == "Keyboard"
db.execute.assert_not_called() # 缓存命中，不查 DB

@pytest.mark.asyncio
async def test_null_cache_prevents_penetration():
service = ProductService()
db = AsyncMock()

with patch.object(cache, 'get', new_callable=AsyncMock) as mock_get:
mock_get.return_value = "__NULL__" # 空值缓存命中

result = await service.get_product(db, 99999)

assert result is None
db.execute.assert_not_called() # 不查 DB

4. 项目总结

优点 & 缺点对比

适用场景

✓ 适用于：

✗ 不适用：

注意事项

常见踩坑经验

案例一：缓存删除失败导致永久脏数据

• 现象：更新了商品价格（DB 中是 99 元），用户看到的还是旧价格 199 元。
• 根因：更新 DB 成功，但 cache.delete() 因为网络抖动失败了——Redis 中仍是旧数据。
• 解决：缓存删除失败时重试 3 次；如果全部失败，写入一个"待删除队列"，由异步任务补偿删除。

案例二：互斥锁死锁

• 现象：热点商品的缓存过期后，所有请求都返回超时——互斥锁的 lock_key 没有被释放。
• 根因：获取锁的请求在执行 finally: delete lock_key 之前进程崩溃了——锁永不过期（什么？设了 5 秒 TTL？哦忘了设）。
• 解决：锁一定要设过期时间（ex=5）。双重保险：在 finally 中释放 + Redis 自动过期兜底。

案例三：大量 key 同时过期的删除风暴

• 现象：整点时刻（如 00:00），Redis 删除 10000 个过期 key，CPU 占用 100%，正常查询的延迟飙升。
• 根因：所有缓存在同一个时间点过期（TTL 相同），Redis 的删除机制集中处理。
• 解决：随机 TTL —— ttl = 3600 + random.randint(0, 600)，每个 key 的过期时间差几百秒，分散删除压力。

思考题

答案提示：第 1 题在 Service 层新增 get_product_list() 方法，先查缓存再查 DB。第 2 题的原理：用 Sorted Set 存储用户请求的时间戳（member=时间戳, score=时间戳），每次请求时清理窗口外的数据，ZCOUNT 统计窗口内请求数。第 22 章深入讲解限流。

延伸阅读与资源

NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化） Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统 Redis 实战修炼与原理进阶 Python 3实战精进：从脚本到高并发订单引擎 python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经 Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地 MongoDB 实战进阶与内核修炼 后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战 10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用 后端工程师转型AI第一课-Ollama 与私有化大模型实战 大型语言模型(LLM) vLLM 高性能推理落地实战 Agent开发之LlamaIndex 实战修炼与源码进阶 大语言模型Transformers 实战修炼与源码剖析