FastAPI+Redis:緩存與狀態管理的基礎應用

在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})防止衝突。

小夜