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