不能直接用 async def 包裹 CPU 密集型任务，因为 GIL 未释放，协程无法让出 CPU，仍会阻塞事件循环；应使用 loop.run_in_executor(ProcessPoolExecutor) 绕过 GIL。

直接用 asyncio 跑 CPU 密集型任务，只会让服务更卡——Python 的 GIL 不会因为你加了 async 就消失。

为什么不能直接用 async def 包裹耗时计算？

异步函数本质是协程调度，适用于 I/O 等待场景；但像矩阵运算、图像处理、复杂解析这类 CPU 绑定操作，会持续占用线程并阻塞事件循环。实测中，一个未释放 GIL 的 async def 函数会让整个 FastAPI/Starlette 服务的并发请求延迟飙升。

• async def calculate_heavy() 内部调用 numpy.linalg.svd() → 仍为同步阻塞
• 即使加了 await asyncio.sleep(0)，也无法让出 CPU 时间片
• 错误日志里不会报错，但 time.perf_counter() 显示单次调用耗时和同步执行完全一致

推荐方案：loop.run_in_executor() + concurrent.futures.ProcessPoolExecutor

这是标准库中唯一无需引入第三方、且能真正绕过 GIL 的方式。关键点在于必须用 ProcessPoolExecutor（不是 ThreadPoolExecutor），否则还是在同一个进程里打转。

• 启动时初始化全局 executor：executor = ProcessPoolExecutor(max_workers=4)
• 在路由中调用：await loop.run_in_executor(executor, heavy_computation, *args)
• 注意传参必须是可序列化的（dataclass、dict、基本类型），不能传 self 或数据库连接对象
• 如果任务需访问 Redis/DB，得在子进程中重新建连接，不能复用主线程的 client 实例

什么时候该换 Celery 或 Dramatiq？

当你的“耗时计算”开始出现以下任意一种情况，就该跳出 run_in_executor：

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

• 单次任务 > 2 分钟，进程池容易因超时被 kill（BrokenProcessPool）
• 需要任务重试、优先级队列、延迟执行（比如“30 秒后发通知”）
• 微服务已拆成多个容器，计算节点要独立扩缩容
• 任务失败后需人工干预或查日志，而进程池日志默认不落盘也不带 trace_id

此时用 Celery 配合 Redis Broker 是最平滑的迁移路径，只需把 run_in_executor 替换为 my_task.delay(*args)，其余逻辑几乎不用动。

别忽略的细节：子进程的模块导入与配置隔离

子进程不会自动继承主进程的 os.environ 或已加载的配置对象。常见翻车点：

• ImportError: No module named 'app.utils' → 子进程工作目录不是项目根目录，需在 heavy_computation 开头加 sys.path.insert(0, "/path/to/project")
• 读取 .env 失败 → 必须显式调用 load_dotenv()，且传入完整路径
• 使用 logging 打印日志却看不到输出 → 子进程 logger 需单独配置 handler，或改用 print()（会被捕获到父进程 stdout）

最稳妥的做法是：所有计算函数保持无状态、参数全显式传入、内部不依赖任何全局变量或单例。

Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！