应使用 redis.asyncio.Redis 配合 ConnectionPool.from_url() 初始化异步连接池,开启 decode_responses=True,设置 max_connections=20、socket_connect_timeout=5、socket_timeout=5 和 retry_on_timeout=True,并传入含密码和DB编号的 Redis URL。

直接结论:用 redis.asyncio.Redis + ConnectionPool.from_url() 初始化连接池,再通过自定义装饰器拦截请求、构造 key、序列化写入,不依赖 fastapi-cache 更可控,尤其在需要按 query、header、用户身份差异化缓存时。
怎么初始化 Redis 连接池(async)
FastAPI 是异步框架,必须用支持 async 的客户端。redis-py 4.0+ 的 redis.asyncio 模块是官方推荐方案,别再用同步的 redis.Redis 或已弃用的 aioredis v1。
关键点:
-
ConnectionPool.from_url()是连接工厂,负责复用连接;直接传 URL 比拆解 host/port 更简洁且兼容密码、DB 编号 -
decode_responses=True必须开启,否则所有get()返回bytes,每次都要手动.decode("utf-8") -
max_connections=20是安全起点,高并发场景下可调到 50–100,但需匹配 Redismaxclients配置 - 超时参数不能省:
socket_connect_timeout=5和socket_timeout=5防止单个慢请求拖垮整个服务
示例代码片段:
import redis.asyncio as redis
pool = redis.ConnectionPool.from_url(
"redis://:mypassword@localhost:6379/0",
max_connections=20,
decode_responses=True,
socket_connect_timeout=5,
socket_timeout=5,
retry_on_timeout=True,
)
redis_client = redis.Redis(connection_pool=pool)
为什么缓存装饰器要自己写,而不是用 fastapi-cache
第三方封装如 fastapi-cache 在简单场景下开箱即用,但一旦涉及以下需求,就会卡住:
- key 要包含
Authorization头哈希值(区分用户级缓存) - 同一接口对
Accept: application/msgpack和application/json返回不同序列化格式 - 需要对 Pydantic v2 的
BaseModel自动调用.model_dump()而非dict() - 缓存未命中时想记录日志或触发指标上报
自己写装饰器核心就三步:生成 key → await redis.get(key) → 命中则反序列化返回,否则执行原函数 + 序列化写入。控制权全在你手上。
缓存 key 怎么构造才不出错
常见错误是 key 只含路径,导致 GET /items?id=100 和 GET /items?id=200 共享同一个缓存,或者不同用户的请求互相污染。
正确做法是组合以下要素并做确定性哈希:
-
request.method(GET/POST 区分) -
request.url.path(不含 query) -
tuple(sorted(request.query_params.items()))(避免str(query_params)顺序不一致) - 可选:对
request.headers.get("authorization")做hashlib.sha256().hexdigest(),用于用户隔离 - 可选:
request.headers.get("accept"),适配多格式响应
最终拼接后走 md5(...).hexdigest() 得到固定长度 key,防止超长 key 触发 Redis 的 key too long 警告。
示例片段:
from hashlib import md5
sorted_q = tuple(sorted(request.query_params.items()))
user_hash = hashlib.sha256(request.headers.get("authorization", "").encode()).hexdigest()
key = f"fastapi:{method}:{path}:{md5((str(sorted_q) + user_hash).encode()).hexdigest()}"
序列化返回值要注意什么
FastAPI 路由函数返回的不是纯 dict,可能是 BaseModel、datetime、UUID、甚至 StreamingResponse。直接 json.dumps() 会报错。
安全做法:
- 统一用
json.dumps(obj, default=str),把不可序列化对象转成字符串(如datetime.isoformat()) - Pydantic v2 推荐先调
obj.model_dump()再 json 序列化,不要json.dumps(obj) - 性能敏感场景可用
msgpack.packb(obj, default=str),体积更小、解析更快,但需客户端支持 msgpack - 绝对不要用
pickle.dumps():有反序列化执行任意代码风险,且跨语言/跨 Python 版本不兼容
缓存降级也得考虑:Redis 连接失败时,应捕获 redis.ConnectionError 或 redis.TimeoutError,跳过缓存直接执行原逻辑,避免雪崩。
最易被忽略的是 TTL 设置粒度——全局统一 CACHE_TTL=300 看似省事,但热点新闻和用户配置项的更新频率完全不同。建议按接口路径分级设置,比如 /api/v1/news 用 60 秒,/api/v1/user/profile 用 3600 秒,并配合主动清理(如用户修改资料后 delete_pattern("fastapi:*user:*"))保障一致性。


















