FastAPI + Redis 实战:一分钟搞定高性能缓存方案
在开发高性能 Web 接口时,我们常常面临这样的问题:
- 某些接口数据变化不频繁,但访问量巨大;
- 某些操作(如排行榜、用户信息)计算/查询代价高昂;
- 数据库访问压力太大,接口响应慢。
这时候,就轮到我们的大杀器——Redis 上场了。
本文带你快速实战:FastAPI 如何优雅接入 Redis 实现缓存与存储!
使用场景:接口缓存
我们以一个用户信息接口为例:根据 user_id 返回用户数据,但用户信息不常变动,可以缓存。
安装依赖
pip install fastapi uvicorn redis
初始化 Redis 客户端
我们使用官方推荐的异步库 redis-py 配合 asyncio 实现异步访问。
# redis_client.py
import redis.asyncio as redis
redis_client = redis.Redis(
host="localhost",
port=6379,
decode_responses=True # 返回字符串而非字节
)
接口代码示例
# main.py
from fastapi import FastAPI, HTTPException
from redis_client import redis_client
import json
app = FastAPI()
# 模拟数据库
fake_user_db = {
"1": {"id": 1, "name": "Alice"},
"2": {"id": 2, "name": "Bob"},
}
@app.get("/user/{user_id}")
async def get_user(user_id: str):
cache_key = f"user:{user_id}"
# 尝试从 Redis 获取缓存
cached_data = await redis_client.get(cache_key)
if cached_data:
return {"source": "redis", "data": json.loads(cached_data)}
# 模拟从数据库查询
user = fake_user_db.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
# 缓存结果,设置过期时间 60 秒
await redis_client.set(cache_key, json.dumps(user), ex=60)
return {"source": "db", "data": user}
说明:
- Redis 作为缓存中间层:查询接口先查 Redis,再查数据库
- 缓存数据使用 JSON 序列化存储
- 通过 ex=60 设置缓存过期时间为 60 秒
Redis 的应用价值
- 降低数据库访问压力
- 提升接口响应速度(缓存命中几乎毫秒级)
- 灵活设置过期时间,控制数据新鲜度
衍生应用场景
- 用户会话 token 存储
- 短信验证码防刷
- 排行榜数据缓存
- 秒杀库存预扣减
- 实现消息队列、分布式锁等
小结
Redis 作为缓存方案在 FastAPI 项目中非常简单易用,也非常高效。它能帮助你:
- 提升接口性能;
- 减轻数据库压力;
- 实现更多灵活应用场景(例如登录状态、验证码、计数器等)。