如何让异步函数在同步上下文中安全运行（asyncio.run）

asyncio.run 不能在已有事件循环中调用，因其设计为启动并关闭全新事件循环；否则抛 RuntimeError。正确做法是：在已有 loop 中用 await、create_task 或 run_until_complete。

asyncio.run 为什么不能在已有事件循环中调用

asyncio.run 的设计目标是启动一个**全新的、干净的事件循环**，执行完协程后立即关闭。如果当前线程已存在运行中的事件循环（比如在 Jupyter、Django shell、或另一个 asyncio.run 内部），再次调用会触发 RuntimeError: asyncio.run() cannot be called from a running event loop。

这不是 bug，而是保护机制：避免嵌套事件循环导致状态混乱、资源泄漏或死锁。

• 常见触发场景：jupyter notebook 中直接写 asyncio.run(my_coro())；Django manage.py shell 里尝试运行异步函数
• 它内部会调用 asyncio.new_event_loop() + loop.run_until_complete() + loop.close()，全程不复用已有 loop
• 即使你手动 asyncio.set_event_loop(None)，也不能绕过该检查 —— asyncio.run 会在开头就调用 asyncio.get_running_loop() 并抛异常

在已有事件循环里怎么安全跑 async 函数

核心原则：别用 asyncio.run，改用 asyncio.create_task() 或 asyncio.ensure_future() 把协程提交给当前 loop；若需同步等待结果，用 asyncio.get_event_loop().run_until_complete()（仅限主线程且 loop 未关闭）。

Magic AI Avatars

神奇的AI头像，获得200多个由AI制作的自定义头像。

下载

• 在 Jupyter / IPython 中，优先用 await my_coro() —— 它们已自动管理事件循环，await 就是正确方式
• 在普通 Python 脚本但已调用过 asyncio.run 后，不能再调 asyncio.run；此时应确保所有异步逻辑都在同一个顶层 asyncio.run 内组织好
• 若必须从同步函数内“插入”异步调用（如 legacy sync API 包装 async 底层），且无法修改调用方为 async，可考虑 asyncio.get_event_loop().run_until_complete(coro)，但要确认 loop 确实正在运行且未关闭

asyncio.run 的替代方案：何时该用 asyncio.new_event_loop

极少数情况需要手动管理事件循环，比如子线程中启动独立异步任务，或测试中隔离环境。这时不用 asyncio.run，而显式创建并管理 loop。

• 子线程中必须调用 asyncio.new_event_loop()，然后 loop.run_until_complete()，最后 loop.close() —— asyncio.run 只在主线程安全
• 手动创建的 loop 不会自动设置为当前线程的默认 loop，需配合 asyncio.set_event_loop(loop)
• 错误示范：loop = asyncio.get_event_loop() 在子线程中会报 RuntimeError: There is no current event loop in thread
• 示例片段：

  import asyncio
  def run_in_thread():
      loop = asyncio.new_event_loop()
      asyncio.set_event_loop(loop)
      try:
          result = loop.run_until_complete(my_coro())
          return result
      finally:
          loop.close()

容易被忽略的陷阱：asyncio.run 的隐式关闭行为

asyncio.run 执行结束后，不仅关闭事件循环，还会取消所有未完成的 task，并清空 loop 的所有回调。这意味着：如果其他代码还持有对这些 task 的引用，或依赖某些 loop-level 的清理逻辑（如 loop.set_exception_handler 注册的处理器），它们可能不会按预期执行。

• 例如：在 asyncio.run 内启动后台 task（asyncio.create_task(...)），但没 await 它，该 task 会被强制 cancel，且其 __aexit__ 或 finally 块可能跳过
• 再如：使用 asyncio.to_thread 启动的线程，若在 asyncio.run 结束前未完成，也会被 abrupt 终止
• 调试时注意：asyncio.run 返回后，asyncio.get_event_loop() 可能返回新 loop，也可能抛 RuntimeError，取决于 Python 版本和是否显式设置了 loop