一、爲什麼請求會超時？¶
想象一下，你點了一份外賣，等了15分鐘還沒送到，系統會提示“配送超時”。同樣，如果你的FastAPI接口處理請求時耗時太長，用戶也會覺得“等得太久”，甚至直接放棄。請求超時的本質是服務器處理請求的時間超過了客戶端的等待閾值，常見原因包括：
- 用戶網絡卡頓：客戶端網絡不穩定，提前終止請求。
- 服務器負載過高：同時有大量請求，服務器處理不過來，導致響應延遲。
- 接口本身耗時：比如調用數據庫查詢、文件讀寫等IO操作，代碼阻塞導致處理時間過長。
超時的危害很明顯：用戶體驗差、請求失敗、甚至引發下游服務連鎖反應。因此，我們需要在FastAPI中主動設置超時，並通過異步處理提升響應速度。
二、FastAPI如何設置請求超時？¶
FastAPI提供了簡單的方式設置請求超時，核心是通過timeout參數控制單個請求的最大處理時間。以下是兩種常見場景：
1. 路由級超時設置¶
在異步路由中，直接在async def函數中設置timeout（單位：秒）。如果請求處理超過該時間，FastAPI會立即返回超時錯誤。
from fastapi import FastAPI
import asyncio
app = FastAPI()
# 模擬耗時15秒的異步任務（比如數據庫查詢、API調用）
@app.get("/slow-task")
async def slow_task(timeout: int = 10): # 設置超時10秒
try:
await asyncio.sleep(15) # 模擬耗時操作
return {"status": "任務完成"}
except TimeoutError:
return {"error": "請求超時！服務器處理時間超過10秒"}
2. 全局超時設置¶
如果希望所有請求統一設置超時，可以通過app的lifespan 或中間件實現。以下是一個簡單的全局超時中間件示例：
from fastapi import Request, Response
from fastapi.responses import JSONResponse
class TimeoutMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
# 設置全局超時爲10秒（注意：需結合異步框架使用）
timeout = 10
try:
return await asyncio.wait_for(
self.app(scope, receive, send),
timeout=timeout
)
except asyncio.TimeoutError:
return JSONResponse(
status_code=504, # 504表示網關超時
content={"error": "全局超時！請稍後重試"}
)
app.add_middleware(TimeoutMiddleware)
三、異步處理：讓FastAPI更快的祕訣¶
爲什麼異步能解決超時問題？同步處理會讓代碼“乾等”，而異步可以在等待時處理其他請求。比如燒水時，同步是“一直盯着水壺等水開”，異步是“設置鬧鐘後去做別的事，水開時再回來”。
1. 同步vs異步：代碼對比¶
- 同步路由（阻塞，適合CPU密集型任務）：
import time
def sync_task():
time.sleep(5) # 同步耗時操作會阻塞整個線程
return "同步完成"
- 異步路由（非阻塞，適合IO密集型任務）：
import asyncio
async def async_task():
await asyncio.sleep(5) # 異步等待，不阻塞線程
return "異步完成"
2. FastAPI中異步路由的寫法¶
FastAPI推薦用async def定義異步路由，配合await調用異步操作。例如：
@app.get("/async-route")
async def async_route():
# 調用異步IO操作（如異步數據庫查詢）
await asyncio.sleep(3) # 模擬異步耗時
return {"message": "異步處理完成"}
3. 異步任務：後臺執行不阻塞¶
如果任務不需要立即返回結果（比如發送郵件、生成報表），可以用BackgroundTasks或asyncio.create_task在後臺執行：
from fastapi import BackgroundTasks
@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks):
# 將耗時任務加入後臺，不阻塞主請求
background_tasks.add_task(send_async, "user@example.com")
return {"status": "郵件已加入發送隊列，立即返回"}
async def send_async(to_email):
await asyncio.sleep(10) # 異步發送郵件（需異步庫支持）
print(f"郵件已發送給{to_email}")
四、性能優化實戰：從代碼到部署¶
優化性能不能只靠超時和異步，還需要從代碼、數據庫、服務器等層面綜合優化。
1. 緩存：減少重複計算¶
對高頻但不常變化的數據（如熱門商品列表），用lru_cache或Redis緩存結果：
from functools import lru_cache
# 緩存最近100次調用結果
@lru_cache(maxsize=100)
def get_cached_data(param: int):
# 實際場景：從數據庫查詢
return param * 2 # 模擬計算
@app.get("/cached-data/{param}")
async def cached_data(param: int):
data = get_cached_data(param)
return {"data": data}
2. 數據庫連接池¶
數據庫操作是最常見的IO瓶頸。使用連接池可以避免頻繁創建/關閉連接：
import asyncpg
# 連接池配置（PostgreSQL示例）
async def init_db_pool():
return await asyncpg.create_pool(
user="user",
password="password",
database="mydb",
host="localhost",
max_size=20 # 最大連接數，根據服務器性能調整
)
# 路由中調用連接池查詢
@app.get("/db-data")
async def db_data(pool=Depends(init_db_pool)):
async with pool.acquire() as connection:
result = await connection.fetchrow("SELECT * FROM users LIMIT 10")
return {"data": result}
3. 數據庫優化¶
- 避免N+1查詢：用
SELECT ... JOIN代替多次單條查詢。 - 索引優化：給高頻查詢字段加索引（如用戶ID、訂單號）。
- 批量操作：多數據寫入時用
INSERT INTO ... VALUES (), ()代替循環單條插入。
4. 部署層面：負載均衡¶
單機性能有限時，通過Nginx、Gunicorn+Uvicorn多進程部署，或使用雲服務的負載均衡：
- Uvicorn多進程：uvicorn main:app --workers 4 --reload（4個工作進程）。
- 負載均衡：將請求分發到多臺服務器（如AWS ELB、阿里雲SLB）。
五、總結：如何綜合優化？¶
- 先解決超時：用
timeout參數設置合理閾值（如10-30秒），避免用戶乾等。 - 異步處理IO密集型任務：所有數據庫、API調用等耗時操作必須用異步庫（如
asyncpg、httpx.AsyncClient）。 - 緩存高頻數據：用
lru_cache或Redis減少重複計算。 - 優化數據庫連接：配置連接池，避免頻繁連接開銷。
- 部署擴展：多進程、負載均衡提升併發處理能力。
通過以上步驟，你的FastAPI接口將既能避免超時，又能高效處理高併發請求，讓用戶體驗和系統穩定性同時提升。