FastAPI + Redis 實戰:一分鐘搞定高性能緩存方案
作者:Ss肥魚
本文帶你快速實戰: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 項目中非常簡單易用,也非常高效。它能幫助你:
- 提升接口性能;
- 減輕數據庫壓力;
- 實現更多靈活應用場景(例如登錄狀態、驗證碼、計數器等)。
責任編輯:趙寧寧
來源:
Ssoul肥魚