如何在一个asyncio事件循环内嵌套使用另一个asyncio事件循环

嘿，这个问题我刚接触asyncio的时候也踩过坑！默认情况下asyncio的设计不允许在一个已运行的事件循环里再启动另一个循环，所以你碰到的RuntimeError: This event loop is already running是非常典型的场景。下面给你几个实用的解决方案，帮你理清思路：

方案1：用asyncio.create_task替代嵌套循环 #

这是最符合asyncio设计理念的做法——不需要启动新的循环，而是把异步任务直接加入到现有循环中。如果你的核心需求是在外部任务里执行内部异步逻辑，完全不需要嵌套循环，直接创建子任务并等待它完成即可。

比如你的基础测试代码可能是这样的错误写法：

import asyncio

async def inner_task():
    print("Inner task running")
    await asyncio.sleep(1)
    print("Inner task done")

async def outer_task():
    print("Outer task running")
    # 错误：直接在现有循环里启动新循环
    asyncio.run(inner_task())
    print("Outer task done")

asyncio.run(outer_task())

改成正确的写法：

import asyncio

async def inner_task():
    print("Inner task running")
    await asyncio.sleep(1)
    print("Inner task done")

async def outer_task():
    print("Outer task running")
    # 正确：将子任务加入现有事件循环
    inner = asyncio.create_task(inner_task())
    await inner  # 等待子任务完成
    print("Outer task done")

asyncio.run(outer_task())

方案2：用nest_asyncio库强制支持嵌套 #

如果因为某些原因（比如维护遗留代码、在Jupyter环境中运行）你必须嵌套循环，可以用nest_asyncio这个第三方库来patch asyncio，让它允许循环嵌套运行。

首先安装库：

pip install nest_asyncio

然后修改代码：

import asyncio
import nest_asyncio

# 打补丁，允许事件循环嵌套
nest_asyncio.apply()

async def inner_task():
    print("Inner task running")
    await asyncio.sleep(1)
    print("Inner task done")

async def outer_task():
    print("Outer task running")
    # 现在可以直接嵌套调用asyncio.run了
    await asyncio.run(inner_task())
    print("Outer task done")

asyncio.run(outer_task())

方案3：手动管理独立的子循环（不推荐用于日常开发） #

如果你想深入理解底层逻辑，也可以手动创建一个独立的子循环，在现有协程里通过线程池来运行它（避免阻塞主循环）。不过这种写法比较繁琐，日常开发优先用前两种方案：

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def inner_task():
    print("Inner task running")
    await asyncio.sleep(1)
    print("Inner task done")

def run_inner_loop():
    # 创建独立的子循环
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(inner_task())
    loop.close()

async def outer_task():
    print("Outer task running")
    # 在线程池里运行子循环，不阻塞主循环
    executor = ThreadPoolExecutor(max_workers=1)
    await asyncio.get_event_loop().run_in_executor(executor, run_inner_loop)
    print("Outer task done")

asyncio.run(outer_task())

为什么会报错？ #

简单来说，asyncio默认要求每个线程只能有一个活跃的事件循环。当你在已运行的循环里调用asyncio.run()时，它会尝试复用当前线程的循环，但这个循环已经处于运行状态，所以就抛出了RuntimeError。

总结一下：优先用方案1，它最符合asyncio的异步编程思想；如果必须嵌套，方案2是最简便的选择；方案3适合用来理解底层机制，但日常开发不推荐。

内容的提问来源于stack exchange，提问作者antfuentes87