run_in_executor需显式配合processpoolexecutor才能解决cpu密集型任务阻塞问题；默认threadpoolexecutor受gil限制无效，且函数须可pickle、进程池应复用、入口需if name == "__main__":防护。

如何在python中使用run_in_executor执行cpu密集计算？

run_in_executor 本身不解决 CPU 密集型任务的阻塞问题

run_in_executor 是 asyncio 提供的“把同步函数扔进线程池或进程池执行”的机制，但它不会自动让 CPU 密集型任务变快或不卡事件循环——关键在于你传给它的 executor 类型。默认用的是 ThreadPoolExecutor，而线程在 Python 中受 GIL 限制，对纯 CPU 计算几乎无效。

如果你发现用了 run_in_executor 后 CPU 计算还是卡住协程、响应延迟高，大概率是因为没显式指定 ProcessPoolExecutor。

必须显式传入 ProcessPoolExecutor 才能真正并行 CPU 计算

asyncio 的 run_in_executor 支持传入自定义 executor。要真正释放多核能力，得用 concurrent.futures.ProcessPoolExecutor，而不是依赖默认线程池。

不要写 loop.run_in_executor(None, cpu_heavy_func, *args) —— None 表示用默认线程池，GIL 下无效
要写 loop.run_in_executor(process_pool, cpu_heavy_func, *args)，其中 process_pool 是提前创建好的 ProcessPoolExecutor 实例
进程池不宜在每次调用时临时创建，否则开销远超计算收益；建议作为全局或类成员复用
被调用的函数（如 cpu_heavy_func）必须可被 pickle，不能是闭包、lambda 或定义在 __main__ 顶层但未加 if __name__ == "__main__": 保护的函数（Windows/macOS 上会报 RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase.）

常见错误：传参失败、结果类型异常、进程启动失败

CPU 密集任务走进程池后，容易在参数传递和返回值上出错，因为跨进程序列化有约束：

所有参数和返回值必须是可 pickle 的：支持 int、str、list、dict、numpy.ndarray（需注意版本兼容），但不支持 threading.Lock、generator、未绑定的实例方法、嵌套的 lambda
如果函数内部打印日志或调用 print()，输出可能不按预期顺序出现，甚至丢失——进程间 stdout 不共享，也不保证 flush
若函数抛出异常，它会被捕获并作为 concurrent.futures.ProcessPoolExecutor 的 BrokenProcessPool 或普通 Exception 抛回协程，需用 try/except 捕获，而不是忽略
在 Jupyter 或某些 IDE 的交互环境中直接运行含 ProcessPoolExecutor 的代码，常因模块导入路径问题失败；建议封装为独立脚本，并确保入口有 if __name__ == "__main__":

一个最小可行示例（含防坑要点）

以下代码能在终端直接运行，绕开了 Windows 多进程典型启动错误，并验证了 CPU 计算是否真正并发：

python全能编程助手

SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的双功能合体技能，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途：；核心要点📝 题解模式：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式：输入 Python 代码，自动添加详细中。它将相关步骤、

下载

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor
from math import factorial
<p>def cpu_intensive(n):</p><h1>纯 CPU 计算，无 IO、无外部依赖</h1><pre class="brush:python;toolbar:false;">return sum(factorial(i) % 1000000 for i in range(n))

async def main():

显式创建进程池，复用而非每次新建

with ProcessPoolExecutor(max_workers=2) as pool:
    loop = asyncio.get_running_loop()
    # 并发提交两个大计算
    t0 = time.time()
    res1 = loop.run_in_executor(pool, cpu_intensive, 10000)
    res2 = loop.run_in_executor(pool, cpu_intensive, 10000)
    await asyncio.gather(res1, res2)
    print(f"Done in {time.time() - t0:.2f}s")

if name == "main": asyncio.run(main())

注意 if __name__ == "__main__": 这一行不是装饰，是 Windows/macOS 多进程必需的防护；with ProcessPoolExecutor(...) 确保资源及时回收；asyncio.gather 等待全部完成，而非逐个 await——否则就退化成串行。

真正麻烦的从来不是怎么写这十几行，而是函数能不能被子进程正确导入、参数有没有隐式携带不可序列化对象、以及错误发生时堆栈指向的是子进程还是主进程——这些细节不排查清楚，run_in_executor 就只是换个方式卡住而已。