在Web開發中,我們經常面臨兩個核心問題:如何快速響應請求,以及如何在多個請求間共享臨時狀態。FastAPI作爲高性能Python Web框架,與Redis(一個內存數據存儲系統)的組合能有效解決這兩個問題。本文將通過簡單示例,帶你理解它們如何協同工作。
一、爲什麼需要FastAPI+Redis?¶
- FastAPI:提供了快速構建API的能力,支持異步請求處理,適合高併發場景。但即使FastAPI再快,對高頻、重複的計算或數據查詢(如用戶信息、統計數據)仍需額外優化。
- Redis:高性能的內存數據庫,支持鍵值對存儲,能快速讀寫數據,且支持過期時間、分佈式鎖等特性。常用於緩存(減少重複計算)和臨時狀態管理(如會話、計數器)。
二、環境準備¶
首先安裝必要的工具:
# 1. 安裝FastAPI和Uvicorn(Web服務器)
pip install fastapi uvicorn
# 2. 安裝Redis Python客戶端
pip install redis
同時確保本地已安裝並啓動Redis服務(默認端口6379,無密碼可直接連接)。驗證連接:
redis-cli # 進入Redis命令行,執行ping,返回PONG則連接成功
三、Redis基礎連接¶
在FastAPI中,我們可以通過工具函數統一管理Redis連接,避免重複創建連接:
from fastapi import FastAPI
import redis
import json
# 初始化Redis連接(生產環境建議用連接池,此處簡化)
redis_client = redis.StrictRedis(
host="localhost", # Redis服務地址
port=6379, # 默認端口
db=0, # 數據庫編號
decode_responses=True # 自動將結果轉爲字符串
)
app = FastAPI()
# 測試連接是否正常
@app.get("/redis-test")
async def redis_test():
redis_client.set("test_key", "Hello Redis!") # 存入鍵值對
return {"message": redis_client.get("test_key")} # 讀取鍵值對
說明:
decode_responses=True確保Redis返回字符串而非字節,方便直接處理。
四、緩存:減少重複計算的利器¶
緩存的核心是將高頻訪問、低更新的數據(如固定配置、用戶基本信息)存入Redis,下次請求直接從緩存讀取,避免重複計算或數據庫查詢。
示例:假設我們有一個API需要“耗時計算”一個結果(如斐波那契數列,模擬複雜邏輯),每次請求都需重新計算,非常低效。用Redis緩存結果:
# 模擬“耗時計算”(實際可能是數據庫查詢或複雜算法)
def slow_calculate(n: int) -> int:
if n <= 1:
return n
return slow_calculate(n-1) + slow_calculate(n-2)
# FastAPI視圖函數:帶緩存邏輯
from fastapi import HTTPException
@app.get("/cache/calculate/{n}")
async def cached_calculate(n: int):
cache_key = f"fib:{n}" # 緩存鍵,用n區分不同計算結果
# 1. 先查緩存
cached_result = redis_client.get(cache_key)
if cached_result:
return {"n": n, "result": int(cached_result), "source": "cache"}
# 2. 緩存未命中,執行耗時計算
try:
result = slow_calculate(n)
except Exception as e:
raise HTTPException(status_code=500, detail="計算失敗")
# 3. 存入緩存,設置過期時間(10分鐘,避免數據永久佔用)
redis_client.setex(cache_key, 600, str(result)) # setex:設置鍵值+過期時間
return {"n": n, "result": result, "source": "database"}
效果:首次請求
/cache/calculate/10會執行計算並緩存,後續請求直接從Redis讀取,無需重複計算。
五、狀態管理:跨請求共享臨時數據¶
Redis也可用於存儲臨時狀態(如用戶會話、操作計數器),解決FastAPI多請求間的狀態共享問題(內存變量僅在單進程內有效,Redis可跨進程/服務器共享)。
示例:統計用戶訪問次數(臨時狀態):
@app.get("/counter/{user_id}")
async def user_counter(user_id: str):
cache_key = f"counter:{user_id}" # 狀態鍵,關聯用戶ID
# 1. 從Redis讀取當前訪問次數
current_count = redis_client.get(cache_key)
if current_count:
current_count = int(current_count) + 1 # 次數+1
else:
current_count = 1 # 新用戶,初始值爲1
# 2. 存入Redis,設置過期時間(24小時,用戶長時間未訪問則自動失效)
redis_client.setex(cache_key, 86400, str(current_count))
return {"user_id": user_id, "訪問次數": current_count}
效果:用戶
user_id=123第一次訪問返回訪問次數=1,第二次訪問返回訪問次數=2,數據在Redis中持久化,即使服務重啓也不會丟失。
六、進階技巧:依賴注入優化代碼¶
爲避免重複編寫Redis連接邏輯,可通過FastAPI的依賴注入統一管理Redis客戶端:
from fastapi import Depends
# 定義Redis連接依賴
async def get_redis_client():
return redis.StrictRedis(host="localhost", port=6379, db=0, decode_responses=True)
# 視圖函數中使用依賴
@app.get("/example")
async def example(redis_client=Depends(get_redis_client)):
redis_client.set("key", "value")
return {"data": redis_client.get("key")}
七、總結¶
- 緩存:核心是“高頻數據→Redis存儲→重複請求直接讀取”,適合減少數據庫/計算壓力(如用戶信息、統計數據)。
- 狀態管理:核心是“跨請求共享臨時數據→Redis鍵值對+過期時間”,適合會話、計數器、權限驗證等場景。
通過FastAPI+Redis的組合,既能發揮FastAPI的異步高性能,又能利用Redis的內存速度優勢,輕鬆解決Web開發中的“響應慢”和“狀態共享”問題。
注意:生產環境需注意Redis的持久化配置(防止數據丟失)、連接池優化(避免頻繁創建連接),以及鍵命名規範(如fib:{n}、counter:{user_id})防止衝突。