Python异步运行同步代码方法

在Python异步编程中，直接在async函数里执行同步CPU密集型或阻塞IO操作会严重拖垮事件循环，导致并发能力归零；正确做法是通过`loop.run_in_executor`将这类任务移交线程池（适合IO等待或轻量CPU工作）或进程池（适合真正多核并行的CPU重载任务），但需严守“传函数对象而非调用结果”“妥善处理异常传播与堆栈定位”“显式管理资源生命周期”等关键细节——忽视这些看似琐碎的实践陷阱，轻则性能不升反降，重则线上服务悄然退化为单线程，调试时却难觅真凶。

为什么不能直接在 async 函数里调用同步 CPU 密集型代码

因为 async 函数运行在事件循环的同一线程中，一旦执行耗时的同步计算（比如 time.sleep(5)、json.loads() 大字符串、数值运算），整个协程就卡住，其他任务无法调度 —— 事件循环被阻塞了，异步就失效了。

常见错误现象：asyncio.sleep() 能挂起，但 time.sleep() 会让整个服务响应变慢；用 requests.get() 发请求后，所有并发请求串行执行；日志里没报错，但吞吐量掉到单线程水平。

• 不是所有同步代码都必须进线程池：小量字符串处理、字典操作、简单条件判断，开销远小于调度成本，直接写就行
• CPU 密集型（如加密、压缩、科学计算）必须进 run_in_executor
• IO 密集型但没异步接口的库（如旧版 requests），也得走 run_in_executor，否则照样阻塞循环

怎么用 loop.run_in_executor 正确包裹同步函数

核心是把同步函数作为参数传进去，别加括号调用；返回的是 asyncio.Future，要 await 才能得到结果。

示例：对一个大列表做 CPU 密集排序

import asyncio
import time
<p>def cpu_heavy_sort(data):
time.sleep(0.1)  # 模拟耗时计算
return sorted(data, reverse=True)</p><p>async def main():
loop = asyncio.get_running_loop()
data = list(range(10000))</p><h1>✅ 正确：传函数对象 + 参数，不调用</h1><pre class="brush:php;toolbar:false"><code>result = await loop.run_in_executor(None, cpu_heavy_sort, data)
print(len(result))</code>

• run_in_executor 第一个参数是 executor，传 None 表示用默认的 ThreadPoolExecutor；生产环境建议显式传入带 max_workers 的实例，避免线程无限增长
• 别写成 await loop.run_in_executor(..., cpu_heavy_sort(data)) —— 这会在主线程先执行完再传结果，完全失去意义
• 如果同步函数需要访问 event loop 或 asyncio 对象（比如想在子线程里再发 async 请求），不行 —— 子线程没有 loop，会报 RuntimeError: There is no current event loop in thread

ThreadPoolExecutor 和 ProcessPoolExecutor 怎么选

Python 的 GIL 让多线程无法真正并行 CPU 计算，所以 ThreadPoolExecutor 对纯 CPU 密集任务提速有限；但对 IO 等待型同步调用（如 subprocess.run()、sqlite3.connect().execute()）足够用，且启动开销小。

真正需要多核并行的 CPU 工作（如矩阵运算、图像处理），得上 ProcessPoolExecutor，但要注意：

• 进程间数据要序列化，函数和参数必须可 pickle；闭包、lambda、类内方法常失败
• 进程启动慢、内存占用高，别为 1ms 计算开新进程
• 不能共享状态：子进程改全局变量、修改传入的 list/dict，父进程看不到
• 异常堆栈可能被截断，调试时记得捕获子进程里的 Exception 并 re-raise 带上下文

容易被忽略的资源清理和异常传播

默认的 ThreadPoolExecutor 不会自动 shutdown，如果脚本长期运行（如 Web 服务），线程可能堆积；更隐蔽的问题是：子线程抛出异常，默认会静默丢失，await 时才冒出来，但堆栈指向 run_in_executor 调用点，不是原始出错行。

• 显式管理 executor：用 async with ThreadPoolExecutor(...) as executor:（需封装适配器）或手动 executor.shutdown(wait=True)
• 在同步函数里加 try/except，把关键信息打日志，或包装成自定义异常类型
• 别依赖 sys.exc_info()：子线程里它拿不到父协程上下文
• 如果同步函数内部用了 atexit 或 signal，注意这些在子线程无效

真正麻烦的从来不是“怎么跑起来”，而是“跑完之后线程还在不在”“错在哪一行根本看不出来”“CPU 占满却没提速”——这些细节不提前压住，上线后查问题的时间远超写代码的时间。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Python异步运行同步代码方法》文章吧，也可关注golang学习网公众号了解相关技术文章。