一、錯誤代碼import asyncio import threading from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI from pydantic import BaseModel app FastAPI() executor ThreadPoolExecutor(max_workers4) global_cache {} cache_lock threading.Lock() # 添加鎖 class ProcessRequest(BaseModel): id: str def cpu_heavy_task(data: ProcessRequest): result 0 for i in range(100000000): result i # 加鎖保護共享緩存 with cache_lock: global_cache[data.id] result return result app.post(/process) async def process_data(data: ProcessRequest): loop asyncio.get_running_loop() result await loop.run_in_executor(executor, cpu_heavy_task, data) return {result: result}二、修改與思考import asyncio from concurrent.futures import ProcessPoolExecutor # 改為進程池 import cachetools # 專業緩存庫 from fastapi import FastAPI, HTTPException app FastAPI() # 進程池數量建議等于 CPU 核心數不要超 executor ProcessPoolExecutor(max_workers4) # 帶 TTL 和最大條目限制的緩存線程安全 cache cachetools.TTLCache(maxsize100, ttl600) def cpu_bound_sync(data_id: str) - int: # 模擬計算注意進程池中無法共享全局變量必須序列化傳入 result sum(range(100000000)) return result app.post(/process) async def process_data(data: ProcessRequest): # 1. 緩存預檢讀緩存無需加鎖TTLCache 內部自帶鎖 if data.id in cache: return {result: cache[data.id], source: cache} # 2. 提交進程池并設置超時防止永久卡死 loop asyncio.get_running_loop() try: result await asyncio.wait_for( loop.run_in_executor(executor, cpu_bound_sync, data.id), timeout30.0 # 超時后任務仍在后臺運行但請求會返回 504 ) except asyncio.TimeoutError: raise HTTPException(status_code504, detailCompute timeout) # 3. 寫入緩存加鎖保護防止多協程并發寫臟數據 # 注ProcessPoolExecutor 返回的結果會 pickle 傳回主進程 cache[data.id] result return {result: result}在微服務架構中絕對不建議讓 API 網關直接等待這種秒級 CPU 計算。更好的做法是用run_in_executor把任務丟進 Redis 隊列如 RQ立即返回202 Accepted和task_id讓獨立的 Worker 進程去計算計算完再回調或輪詢結果。這時候run_in_executor僅用于把任務入隊毫秒級 I/O根本不會堵塞線程池。