FastAPI: 缓存 — Redis 加速热点查询
最后更新：2026-08-26
缓存像冰箱——常吃的食材（热点数据）放在冰箱里随手可得，不用每次都去超市（数据库）。但食材会过期（TTL），换了新货要更新（缓存失效）。
1. 你将学到
- Redis 连接管理：
redis-py异步客户端与 FastAPI 集成 - 缓存策略：Cache-Aside 模式、TTL 过期、缓存键命名设计
- 缓存失效：写操作时主动失效、批量价格更新时的缓存一致性
- 缓存穿透/击穿/雪崩防护：布隆过滤器、互斥锁、随机 TTL
- Alice 场景：热门商品价格缓存——P99 延迟从 200ms 降至 5ms
2. Alice 的真实故事
(1) 痛点：热门查询压垮数据库
PriceTracker 有 100 个热门商品占查询量的 80%，每秒被查询 8000 次。每次查询都要走 PostgreSQL，数据库 CPU 飙到 90%，P99 延迟 200ms。Charlie 说加数据库实例要 USD 500/月，但 80% 的查询结果 5 分钟内不会变。
(2) Redis 缓存的解法
用 Redis 缓存热门商品价格数据，查询先查 Redis（5ms），未命中再查 PostgreSQL（50ms）并写入 Redis，设置 5 分钟 TTL。
PYTHON
async def get_product_cached(product_id: int, db, redis):
cached = await redis.get(f"product:{product_id}")
if cached:
return json.loads(cached)
product = await db.execute(select(Product).where(Product.id == product_id))
data = product.scalar_one_or_none()
await redis.setex(f"product:{product_id}", 300, json.dumps(data))
return data
(3) 收益
热门查询 80% 命中 Redis 缓存，P99 延迟从 200ms 降到 5ms，数据库 QPS 从 8000 降到 1600，Charlie 的数据库实例从 4 个降到 1 个，月省 USD 1500。
3. Redis 连接管理
(1) 异步客户端集成
▶ 示例：Redis 连接依赖
PYTHON
import redis.asyncio as aioredis
from fastapi import FastAPI, Depends
REDIS_URL = "redis://localhost:6379/0"
# Application-level Redis connection pool
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)
async def get_redis() -> aioredis.Redis:
"""Request-scoped Redis client from connection pool"""
client = aioredis.Redis(connection_pool=redis_pool)
try:
yield client
finally:
await client.aclose()
# Startup/shutdown lifecycle
app = FastAPI()
@app.on_event("startup")
async def startup():
global redis_pool
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)
@app.on_event("shutdown")
async def shutdown():
await redis_pool.disconnect()
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) Redis 数据结构选型
| 结构 | 命令 | 适用场景 | PriceTracker 用例 |
|---|---|---|---|
| String | SET/GET | 单值缓存 | 商品价格缓存 |
| Hash | HSET/HGET | 对象缓存 | 商品详情缓存 |
| List | LPUSH/LRANGE | 时间序列 | 价格变更日志 |
| Set | SADD/SMEMBERS | 去重集合 | 已推送价格 ID |
| ZSet | ZADD/ZRANGE | 排行榜 | 热门商品排行 |
4. Cache-Aside 模式
(1) 读写流程
flowchart TD
Request[Read Request] --> CheckRedis{Cache Hit?}
CheckRedis -->|Yes| Return[Return Cached Data]
CheckRedis -->|No| QueryDB[Query Database]
QueryDB --> WriteCache[Write to Redis with TTL]
WriteCache --> Return2[Return Data]
WriteRequest[Write Request] --> UpdateDB[Update Database]
UpdateDB --> Invalidate[Invalidate Cache]
Invalidate --> Return3[Return Success]
▶ 示例：Cache-Aside 商品查询
PYTHON
import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis
app = FastAPI()
@app.get("/api/v1/products/{product_id}")
async def get_product(
product_id: int,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
):
# Step 1: Check Redis cache
cache_key = f"product:{product_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# Step 2: Cache miss - query database
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Step 3: Write to cache with TTL
product_data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(product_data)) # 5 min TTL
return product_data
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：写操作时主动失效
PYTHON
@app.put("/api/v1/products/{product_id}")
async def update_product(
product_id: int,
update: ProductUpdate,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
user=Depends(get_current_user),
):
repo = ProductRepository(db)
product = await repo.update(product_id, update.model_dump(exclude_unset=True))
# Invalidate cache after write
await redis.delete(f"product:{product_id}")
return ProductResponse.model_validate(product).model_dump()
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 缓存键命名规范
| 模式 | 锓名 | TTL | 说明 |
|---|---|---|---|
| 实体缓存 | product:{id} |
5 min | 单商品详情 |
| 列表缓存 | products:cat:{category}:p:{page} |
2 min | 分页列表 |
| 计数器 | count:products:cat:{category} |
1 min | 商品计数 |
| 热门排行 | ranking:products:hot |
10 min | ZSet 排行 |
| 限流计数 | ratelimit:{ip} |
60 s | API 限流 |
5. 缓存穿透/击穿/雪崩防护
(1) 三大缓存问题
| 问题 | 原因 | 防护方案 |
|---|---|---|
| 穿透 | 查不存在的数据，绕过缓存直达 DB | 布隆过滤器、缓存空值 |
| 击穿 | 热点 Key 过期瞬间大量请求直达 DB | 互斥锁、永不过期+异步更新 |
| 雪崩 | 大量 Key 同时过期 | 随机 TTL 偏移 |
▶ 示例：缓存空值防穿透
PYTHON
@app.get("/api/v1/products/{product_id}")
async def get_product_with_null_cache(
product_id: int,
db: AsyncSession = Depends(get_db),
redis: Redis = Depends(get_redis),
):
cache_key = f"product:{product_id}"
cached = await redis.get(cache_key)
if cached:
data = json.loads(cached)
if data.get("_null"):
raise HTTPException(status_code=404, detail="Product not found")
return data
# Query DB
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if not product:
# Cache null value with short TTL to prevent cache penetration
await redis.setex(cache_key, 60, json.dumps({"_null": True}))
raise HTTPException(status_code=404, detail="Product not found")
product_data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(product_data))
return product_data
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：互斥锁防击穿
PYTHON
import asyncio
async def get_product_with_lock(
product_id: int,
db: AsyncSession,
redis: Redis,
):
cache_key = f"product:{product_id}"
lock_key = f"lock:product:{product_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
# Try to acquire lock (only one request rebuilds cache)
lock_acquired = await redis.set(lock_key, "1", nx=True, ex=10)
if lock_acquired:
try:
# This request rebuilds the cache
repo = ProductRepository(db)
product = await repo.get_by_id(product_id)
if product:
data = ProductResponse.model_validate(product).model_dump()
await redis.setex(cache_key, 300, json.dumps(data))
return data
finally:
await redis.delete(lock_key)
else:
# Other requests wait briefly and retry cache
await asyncio.sleep(0.1)
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
raise HTTPException(status_code=404, detail="Product not found")
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：随机 TTL 防雪崩
PYTHON
import random
def get_cache_ttl(base_ttl: int = 300, jitter: int = 60) -> int:
"""Add random jitter to TTL to prevent cache avalanche"""
return base_ttl + random.randint(-jitter, jitter)
# Usage
await redis.setex(cache_key, get_cache_ttl(300, 60), json.dumps(data))
# TTL: 240-360 seconds (5 min ± 1 min jitter)
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 缓存一致性策略对比
| 策略 | 写操作 | 优点 | 缺点 |
|---|---|---|---|
| Cache-Aside | 先更新 DB，再删缓存 | 简单可靠 | 短暂不一致窗口 |
| Write-Through | 同步写缓存和 DB | 一致性好 | 写延迟高 |
| Write-Behind | 先写缓存，异步写 DB | 写性能好 | 可能丢数据 |
❓ 常见问题
Q 缓存应该存 JSON 还是 Pickle？
A 推荐 JSON。可读、跨语言、安全（Pickle 有反序列化漏洞）。用 orjson 比 json 快 3-10 倍。
Q TTL 设多久合适？
A 取决于数据变更频率。PriceTracker 价格数据 5 分钟（允许短暂不一致），商品详情 30 分钟，排行榜 10 分钟。加随机偏移防雪崩。
Q Redis 连接池大小怎么设？
A 与 FastAPI Worker 数量匹配。每个 Worker 需 5-10 个 Redis 连接。4 Worker × 10 = 40 连接，设 max_connections=50 留余量。
Q 布隆过滤器怎么实现？
A 用
redisbloom 模块或 Python pybloom_live 库。所有存在的商品 ID 加入布隆过滤器，查询前先检查。误判率约 1%，可接受。Q 缓存和数据库不一致怎么办？
A Cache-Aside 模式下，短暂不一致是正常的。策略：先更新 DB 再删缓存（而非更新缓存），配合延迟双删（删缓存 → 更新 DB → 延迟再删）减少不一致窗口。
Q 如何监控缓存命中率？
A Redis 的
INFO stats 命令返回 keyspace_hits 和 keyspace_misses，命中率 = hits / (hits + misses)。集成到 Prometheus + Grafana 监控面板。📖 小节
- Redis 异步客户端（
redis.asyncio）+ 连接池 + yield 依赖注入集成 FastAPI - Cache-Aside 模式：读先查缓存，未命中查 DB 并写入缓存；写时更新 DB 并删除缓存
- 缓存键命名遵循
{entity}:{id}模式，不同场景设不同 TTL - 空值缓存防穿透、互斥锁防击穿、随机 TTL 防雪崩——三大防护策略
- PriceTracker 热门商品缓存命中率 80%，P99 延迟从 200ms 降到 5ms
📝 作业
- 基础题（难度⭐）：配置 Redis 异步客户端依赖，实现简单的 GET/SET 缓存端点，验证数据可以存入和读取。提示：
redis.asyncio+Depends(get_redis)+setex() - 进阶题（难度⭐⭐）：为 PriceTracker 商品查询实现 Cache-Aside 模式：读查缓存→未命中查 DB→写缓存（TTL 5 分钟），写操作更新 DB 后删除缓存。提示：
await redis.get(key)+await redis.delete(key) - 挑战题（难度⭐⭐⭐）：实现完整的缓存防护——空值缓存防穿透（短 TTL 60s）、互斥锁防击穿（
SET lock NX EX）、随机 TTL 防雪崩（300s ± 60s），并编写测试验证防护逻辑正确。提示：redis.set(lock_key, "1", nx=True, ex=10)+random.randint()
---|
