一、错误代码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根本不会堵塞线程池。
一、错误代码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根本不会堵塞线程池。
概念介绍)
