第一个通宵：Jupyter Notebook 里运行直接报错

事情要从一个周一早上说起。

数据同事急匆匆跑来找我，说训练脚本在 Jupyter Notebook 里无法运行，直接抛出了一个典型的 Python 异步错误：

RuntimeError: asyncio.run() cannot be called from a running event loop

我第一时间检查了代码，逻辑本身并没有问题——就是很标准的 asyncio 用法：

async def fetch_data():
# 从数据库异步拉取训练数据
return await db.query("SELECT * FROM training_set")

def load_training_data():
data = asyncio.run(fetch_data())
return data

这段代码在命令行环境里运行得很正常，但一放进 Jupyter Notebook 就立刻崩了。

折腾了很久，最后发现原因其实很简单：Jupyter Notebook 本身就已经启动了一个事件循环。

asyncio.run() 的底层机制是：创建一个新的事件循环，执行协程，执行结束后再把这个循环关闭。

但在 Jupyter 里，已经存在一个正在运行的事件循环——asyncio.run() 检测到“当前线程中已有事件循环正在运行”，就会直接拒绝执行。

你可以把它理解成：餐厅里已经有一个服务员在工作了，你还非要再安排一个同岗位的人上岗——系统当然不允许，因为一个线程里同一时间只能有一个事件循环。

解决办法是什么？在 Jupyter Notebook 里直接使用 await，不要再调用 asyncio.run()：

# 在 Jupyter 里直接这样写
data = await fetch_data()

第一个通宵就这样过去了。当时我心里想：好吧，asyncio.run() 不能在已有事件循环的环境中调用，这条规则算是记住了。

但没想到，这只是开始。

第二个通宵：事件循环已关闭

两周之后，我接手了一个数据处理流水线，需要顺序执行三个异步任务：

async def step1(): ...
async def step2(): ...
async def step3(): ...

# 当时的写法
asyncio.run(step1())
asyncio.run(step2())
asyncio.run(step3())

结果 step1 跑完后，执行 step2 时直接报错：

RuntimeError: Event loop is closed

这下就有点懵了。每个 step 单独执行都没问题，为什么串起来就出错？

后来翻源码才明白：**asyncio.run() 每次调用，都会经历“创建事件循环 → 运行协程 → 关闭事件循环”的完整生命周期**。

第一次执行 asyncio.run(step1()) 结束后，事件循环已经被关闭。第二次再调用时，它虽然会尝试重新创建新循环，但某些场景下之前关闭后的状态会引发问题。

asyncio.run() 这个 API 的设计初衷，就是作为整个异步程序的统一入口，在程序顶层调用一次。它并不是为反复调用而设计的。

正确写法应该是：

async def main():
await step1()
await step2()
await step3()

asyncio.run(main()) # 只调用一次

或者，如果你确实需要在同一个事件循环中执行多个独立协程任务，可以使用 loop.run_until_complete()：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(step1())
loop.run_until_complete(step2())
loop.run_until_complete(step3())
finally:
loop.close()

run_until_complete() 不会自动关闭事件循环，因此可以在同一个循环里多次调用。不过要注意，循环的创建与关闭都需要你自己管理——如果忘了 close()，就可能造成资源泄漏。

第二个通宵，我又记住了一条关键经验：**asyncio.run() 通常应该只作为主入口调用一次。**

第三个通宵：在已有事件循环里调用 run_until_complete

又过了两周，代码终于上线了。在一个已经运行中的异步服务里，我调用了第三方库的一个同步封装函数。

那个函数内部是这样写的：

# 第三方库的代码
def do_something_sync():
loop = asyncio.get_event_loop()
loop.run_until_complete(some_async_task())

服务本来就运行在事件循环之中，调用这个函数后马上报错——

RuntimeError: This event loop is already running

run_until_complete() 的使用规则很明确：它只能在事件循环尚未运行时调用。如果当前循环已经在跑，再执行 run_until_complete() 就一定会发生冲突。

这就像车已经启动了，你还非要再拧一次点火钥匙。

那正确做法是什么？如果事件循环已经在运行，就应该使用 create_task() 或直接 await，而不是再次调用 run_until_complete()。

# 在已经运行的事件循环里
task = asyncio.create_task(some_async_task())
# 或者直接 await
await some_async_task()

第三个通宵让我彻底明白：**run_until_complete() 也不能在已有运行中事件循环的环境里随意调用。**

所以到底有什么区别？

折腾了三个通宵之后，我终于把 asyncio.run() 和 loop.run_until_complete() 的区别彻底搞清楚了。

asyncio.run() —— 一站式全包方案

创建一个全新的事件循环
运行你传入的协程对象
执行完成后关闭循环，并清理相关资源（如异步生成器、线程池等）
缺点：每次调用都会新建循环，不能在已有事件循环环境中调用，也不适合频繁重复调用
适用场景：Python 异步脚本的主入口，而且通常只调用一次

loop.run_until_complete() —— 更底层的手动方案

在已有的事件循环上执行协程
执行结束后不会自动关闭循环
缺点：需要自己管理事件循环的创建、设置和关闭；如果循环已经在运行，则不能再调用
适用场景：需要复用同一个事件循环执行多个任务，或者需要手动控制事件循环生命周期

一张图看清楚：

源码里其实早就藏着答案

看一眼 asyncio.run() 的简化源码，逻辑会非常清楚：

def run(main, *, debug=False):
# 检查是否已有运行中的循环——有就报错
if events._get_running_loop() is not None:
raise RuntimeError("asyncio.run() cannot be called from a running event loop")

# 创建全新的循环
loop = events.new_event_loop()
try:
events.set_event_loop(loop)
# 本质上调用的就是 run_until_complete
return loop.run_until_complete(main)
finally:
# 清理：取消所有任务、关闭异步生成器、关闭循环
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()

看到这里就很明白了：**asyncio.run() 的底层核心其实也是调用 loop.run_until_complete()**，只是它在外层额外封装了“创建事件循环 + 清理资源 + 关闭循环”这套完整流程。

asyncio.run() 存在的意义，就是帮开发者把“创建循环、执行协程、关闭循环”这三件事统一打包，减少样板代码，提高可读性。

但这种打包方式的代价也很明显——你失去了对事件循环生命周期的控制。想复用事件循环？不行。想在 Jupyter 或已有异步服务中再调用？也不行。

实战：什么时候该用什么？

场景一：编写独立的 Python 异步脚本

优先使用 asyncio.run()，调用一次即可，代码最简洁：

async def main():
# 你的异步逻辑
pass

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

场景二：在 Jupyter / IPython 中调试异步代码

不要使用 asyncio.run()，直接写 await 更合适：

# Jupyter 里直接 await
result = await my_async_function()

场景三：需要顺序执行多个独立任务，并复用同一个事件循环

可以采用手动管理事件循环的方式：

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(task1())
loop.run_until_complete(task2())
loop.run_until_complete(task3())
finally:
loop.close()

或者使用 Python 3.11+ 提供的 asyncio.Runner：

with asyncio.Runner() as runner:
runner.run(task1())
runner.run(task2())
runner.run(task3())

场景四：在一个异步函数内部调用另一个异步函数

最直接、最正确的方式就是用 await：

async def main():
result = await another_async_function()

三个通宵换来的教训

第一个通宵：asyncio.run() 不能在已有事件循环的环境中调用。

第二个通宵：asyncio.run() 更适合作为程序主入口，一般只调用一次。

第三个通宵：run_until_complete() 同样不能在已经运行的事件循环里调用。

这三个问题，本质上都指向同一个核心原则：

事件循环是单线程中的唯一调度中心，一个线程在同一时刻只能有一个正在运行的事件循环。

asyncio.run() 会替你管理这个事件循环——负责创建、执行和关闭。你要么完全交给它处理（主入口调用一次），要么就自己接管（使用 loop.run_until_complete() 并手动管理生命周期）。两种方式不要混着来。

现在回头看，这三个通宵的经验其实可以总结成一句很适合记忆的话：

asyncio.run() 是帮你创建并管理事件循环的“一站式方案”；loop.run_until_complete() 是让你在已有循环上执行任务的“底层工具”。前者简单省事，但限制更多；后者更灵活，但也意味着你要承担更多管理责任。

到底该选哪个？取决于你的具体开发场景。但无论怎么选，都要记住一条非常重要的规则：

不要在已经运行的事件循环里，再去启动另一个事件循环。

这条规则已经救过我好几次。希望也能帮你避开同样的坑。