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

在同步代码中调用协程，核心思路是**启动事件循环并运行协程直到完成**。Python 不允许直接执行协程对象（如 my_coro() 返回的是协程对象，不是结果），必须通过事件循环驱动。最常用、最安全的方式是使用 asyncio.run()（Python 3.7+）。

用 asyncio.run() 直接运行协程

这是最推荐的方案，适用于顶层同步入口（如脚本主逻辑、命令行工具、测试函数等）。它会自动创建新事件循环、运行协程、等待完成、清理资源。

• 只能调用一次 —— asyncio.run() 会关闭事件循环，重复调用会报错
• 不能在已有事件循环中调用（例如在 Jupyter、Django 视图、或已运行的 asyncio 程序里）
• 适合“从同步进入异步”的一次性场景

示例：

import asyncio
async 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()。此时应获取当前循环并运行协程。

• 检查是否已有运行中的循环：asyncio.get_event_loop() 或 asyncio.get_running_loop()（后者更严格，只返回正在运行的）
• 确保不嵌套调用 —— 同步函数被协程调用时再调此方法会导致 RuntimeError
• 常见于调试、交互式环境或混合架构中临时触发异步操作

示例：

import asyncio
async def hello():
return "Hello from async!"
def sync_call_in_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
l
oop = 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 实例。