同步代码调用协程需通过事件循环驱动,推荐Python 3.7+用asyncio.run()启动新循环并运行协程;已有事件循环时用run_until_complete();多线程中须为子线程单独创建并设置事件循环。

在同步代码中调用协程,核心思路是**启动事件循环并运行协程直到完成**。Python 不允许直接执行协程对象(如 my_coro() 返回的是协程对象,不是结果),必须通过事件循环驱动。最常用、最安全的方式是使用 asyncio.run()(Python 3.7+)。
用 asyncio.run() 直接运行协程
这是最推荐的方案,适用于顶层同步入口(如脚本主逻辑、命令行工具、测试函数等)。它会自动创建新事件循环、运行协程、等待完成、清理资源。
- 只能调用一次 ——
asyncio.run()会关闭事件循环,重复调用会报错 - 不能在已有事件循环中调用(例如在 Jupyter、Django 视图、或已运行的 asyncio 程序里)
- 适合“从同步进入异步”的一次性场景
示例:
import asyncioasync def fetch_data(): await asyncio.sleep(1) return "done"
✅ 正确:在同步函数中调用
def sync_func(): result = asyncio.run(fetch_data()) # 阻塞直到协程完成 print(result) # 输出:done
sync_func()
在已有事件循环中用 asyncio.get_event_loop().run_until_complete()
当你的程序已经运行着一个事件循环(比如你正在写一个 asyncio 服务,或在 Jupyter 中),就不能再用 asyncio.run()。此时应获取当前循环并运行协程。
立即学习“Python免费学习笔记(深入)”;
免费 盛世企业网站管理系统(SnSee)系统完全免费使用,无任何功能模块使用限制,在使用过程中如遇到相关问题可以去官方论坛参与讨论。开源 系统Web代码完全开源,在您使用过程中可以根据自已实际情况加以调整或修改,完全可以满足您的需求。强大且灵活 独创的多语言功能,可以直接在后台自由设定语言版本,其语言版本不限数量,可根据自已需要进行任意设置;系统各模块可在后台自由设置及开启;强大且适用的后台管理支
- 检查是否已有运行中的循环:
asyncio.get_event_loop()或asyncio.get_running_loop()(后者更严格,只返回正在运行的) - 确保不嵌套调用 —— 同步函数被协程调用时再调此方法会导致 RuntimeError
- 常见于调试、交互式环境或混合架构中临时触发异步操作
示例:
import asyncioasync def hello(): return "Hello from async!"
def sync_call_in_loop(): try: loop = asyncio.get_running_loop() except RuntimeError: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) result = loop.run_until_complete(hello()) print(result)
sync_call_in_loop()
避免在多线程中误用事件循环
每个线程默认没有事件循环。如果你在子线程中调用协程,必须手动创建并设置循环;主线程的循环对子线程不可见。
- 子线程中:先调
asyncio.new_event_loop(),再set_event_loop(),然后run_until_complete() - 不要跨线程共享同一个 loop(非线程安全)
- 更推荐用
asyncio.to_thread()(Python 3.9+)把阻塞操作移到线程池,而不是反向把协程拉进线程
不推荐的做法:自建循环 + run_forever + stop
手动调用 loop.run_forever() 并配合 loop.stop() 容易出错,需手动管理生命周期、回调、异常传播,且难以保证 clean shutdown。除非你正在实现底层框架,否则绕开它。
本质上,同步代码调用协程不是“转换执行模型”,而是“委托给事件循环执行并等待结果”。选对方式的关键是看所处上下文:全新环境用 run(),已有 loop 用 run_until_complete(),多线程注意隔离 loop 实例。









