asyncio.run 是做什么的？事件循环生命周期应该怎么管理？

简化版

asyncio.run(coro) 是运行异步程序的推荐入口，它会创建事件循环、运行传入协程、等待异步生成器和默认 executor 清理，最后关闭事件循环。它不能在已经运行的事件循环里再次调用；在 Web 框架、Jupyter、异步服务中，事件循环通常由宿主环境管理，不应该自己随便 asyncio.run()。

详细版

普通脚本里推荐这样启动异步程序：

import asyncio

async def main():
    ...

if __name__ == "__main__":
    asyncio.run(main())

asyncio.run() 负责完整生命周期，比手动 get_event_loop().run_until_complete() 更安全。它适合程序最外层入口，不适合在已有事件循环内部嵌套调用。比如 FastAPI 视图、Jupyter Notebook、异步测试框架里，事件循环已经在运行，此时再调用 asyncio.run() 会报错。

面试回答时要区分三个层次：协程对象只是“可等待的任务描述”；事件循环负责调度；asyncio.run() 是脚本入口层的生命周期管理工具。

完整版教学

一、为什么需要一个统一入口

async def 调用后不会立刻执行函数体，而是返回一个协程对象。这个对象需要被事件循环驱动，里面的 await 才会一步步向前运行。初学者常见错误是只调用协程却没有等待它。

async def main():
    print("hello")

main()  # 只创建协程对象，不会真正打印 hello

正确入口是：

asyncio.run(main())

这相当于告诉 Python：创建一个事件循环，把 main() 作为顶层协程跑完，并在结束时清理资源。对于命令行脚本、定时任务、小工具，这就是最推荐的启动方式。

二、asyncio.run 内部大致做了什么

asyncio.run() 不是简单调用 run_until_complete()。它会创建新的事件循环，设置为当前上下文的运行循环，执行顶层协程，处理异步生成器关闭和默认线程池 executor 清理，最后关闭 loop。

可以简化理解成：

asyncio.run(main())
  |
创建新 event loop
  |
运行 main 协程直到完成
  |
清理异步生成器 / executor
  |
关闭 event loop

这个封装减少了手动管理 loop 的遗漏。早期代码里经常能看到 loop = asyncio.get_event_loop(); loop.run_until_complete(...)，但现代脚本入口优先用 asyncio.run()，除非你确实需要自定义 loop 策略或嵌入到框架里。

三、为什么不能嵌套 asyncio.run

事件循环本身就是调度器，一个线程里通常不应该同时跑两个嵌套的 asyncio 事件循环。如果当前线程已经有 loop 正在运行，再调用 asyncio.run() 创建并运行另一个 loop，会破坏调度模型，所以会报错。

典型错误：

async def handler():
    asyncio.run(do_work())  # RuntimeError

正确做法是在异步函数里直接 await：

async def handler():
    await do_work()

数字化理解：外层事件循环正在管理 100 个连接，如果某个连接处理函数里又阻塞式启动另一个 loop，外层 loop 无法正常调度其他连接。异步代码要把控制权交还给当前 loop，而不是另起炉灶。

在 async 函数里看到 asyncio.run() 基本要警觉：多数情况下应该改成 await 或 create_task()。

四、不同环境下谁管理事件循环

脚本里你管理事件循环；框架里框架管理事件循环。FastAPI/Uvicorn、aiohttp、异步测试插件、Jupyter Notebook 都会在外层创建并运行 loop，你只需要写 async 函数并 await 该 await 的对象。

环境	loop 谁管理	你的写法
命令行脚本	自己	`asyncio.run(main())`
FastAPI 接口	ASGI Server	`async def` 里 `await`
Jupyter	Notebook 内核	顶层可直接 `await`
pytest-asyncio	测试插件	`async def test_xxx()`
库函数	调用方	返回协程或提供 async API

库代码尤其不要随便调用 asyncio.run()。如果一个库函数内部强行创建事件循环，它会让已经运行在异步环境中的调用方无法使用。更好的方式是提供 async 函数，让应用入口决定怎么运行。

五、协程、Task 和事件循环生命周期的关系

协程对象本身不会并发执行；被包装成 Task 后，事件循环才能调度它和其他任务交替运行。asyncio.run(main()) 只负责顶层入口，main() 内部可以创建多个任务并等待它们完成。

async def main():
    t1 = asyncio.create_task(fetch(1))
    t2 = asyncio.create_task(fetch(2))
    await t1
    await t2

如果 main() 返回时还有未等待的后台任务，它们可能被取消或留下警告。工程上要明确任务归属：创建了任务就要等待、取消、收集异常，不能随手 create_task() 后完全不管。

main 开始
  |
创建 task A / task B
  |
await gather(A, B)
  |
main 结束 -> asyncio.run 清理 loop

六、手动管理 loop 什么时候还有必要

多数应用脚本不需要手动管理 loop。但在嵌入式运行、特殊 loop 策略、需要长期运行服务、与 GUI/其他事件系统集成、底层框架实现中，仍可能看到手动创建和关闭 loop。

loop = asyncio.new_event_loop()
try:
    asyncio.set_event_loop(loop)
    loop.run_until_complete(main())
finally:
    loop.close()

这种写法要自己负责清理细节，容易遗漏异步生成器、executor、未完成任务。面试里可以说：应用入口优先用 asyncio.run()；框架和高级场景才手动管 loop，且要非常清楚生命周期边界。

七、常见误区与追问

误区：调用 async 函数就会自动执行。 调用只返回协程对象，必须 await 或交给事件循环运行。
误区：哪里都可以用 asyncio.run()。 它是最外层入口工具，不能在已运行事件循环中嵌套调用。
误区：create_task() 后任务一定安全完成。 如果没有等待和异常处理，任务可能被取消，异常也可能无人处理。
追问：FastAPI 里为什么不能 asyncio.run()？ Uvicorn 已经运行事件循环，接口函数在这个 loop 内执行，应直接 await。
追问：Jupyter 里为什么可以直接 await？ Notebook 内核已经提供运行中的事件循环和顶层 await 支持。
追问：库函数应不应该调用 asyncio.run()？ 通常不应该；库应暴露 async API，让调用方决定运行环境。
追问：asyncio.run() 结束后 loop 还能复用吗？ 不能，它会关闭自己创建的事件循环。

八、加强记忆

把 asyncio.run() 记成“脚本入口的一次性 loop 管家”：它创建 loop，跑完顶层协程，清理资源，再关闭 loop。async 函数内部不要再开新管家，直接 await 当前 loop 里的工作；框架环境也别抢 loop 的管理权。理解“协程是描述，Task 是调度单元，loop 是调度器，run 是入口生命周期”这四层，异步启动问题就不会乱。

简化版

详细版