run_in_executor需显式配合ProcessPoolExecutor才能解决CPU密集型任务阻塞问题；默认ThreadPoolExecutor受GIL限制无效，且函数须可pickle、进程池应复用、入口需if name == "__main__":防护。
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 密集任务走进程池后，容易在参数传递和返回值上出错，因为跨进程序列化有约束：
立即学习“Python免费学习笔记（深入）”；
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
- 所有参数和返回值必须是可 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 计算是否真正并发：
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 就只是换个方式卡住而已。
