Python asyncio 事件循环嵌套原理、陷阱与三种解法写这篇笔记的起因是最近在一个 FastAPI 服务里集成异步推理脚本时反复踩到RuntimeError: This event loop is already running这个坑。表面上看只是一个报错但解决它的过程中发现如果不理解事件循环的底层机制三种常见的修补方案选错一个后续就会埋下一串隐蔽的 bug。这篇笔记把原理和工程实践串一遍方便以后回头查。一、事件循环的实质单线程内的任务调度器Python 的asyncio不是多线程也不是多进程。它的并发模型叫做cooperative multitasking协作式多任务——所有协程coroutine跑在同一个线程里通过主动让出控制权来实现看起来像同时做很多事的效果。事件循环Event Loop就是这个调度中枢。它的工作逻辑可以简化为一个无限循环# 伪代码描述事件循环的核心逻辑loopSelectorEventLoop()whilenotloop._stopping:# 1. 执行所有到期的回调scheduled callbacksloop._run_scheduled_callbacks()# 2. 检查 IO 多路复用的就绪状态timeoutloop._compute_next_timeout()eventsselector.select(timeout)# 3. 把 IO 就绪的协程重新放入可执行队列forkey,maskinevents:loop._process_events(key,mask)# 4. 执行就绪队列中的协程直到它们再次 awaitwhileloop._ready:handleloop._ready.popleft()handle._run()几个不能混淆的概念协程coroutine由async def定义的函数调用时返回一个协程对象本身不会执行任务Task协程对象的包装器被注册到事件循环的队列里等待调度。asyncio.create_task()就是干这个的事件循环Event Loop唯一的调度器负责维护任务队列、监听 IO 事件、唤醒挂起的协程关键约束一个线程同一时刻只能有一个事件循环在运行。这不是性能限制而是设计约束——如果同一个线程里有两个循环同时抢任务队列回调的执行顺序就不可预期了。二、嵌套问题的根源asyncio.run()的防御性检查asyncio.run()是 Python 3.7 引入的便利函数它的源码非常简短核心逻辑只有三步# asyncio/runners.py Python 3.11简化版defrun(main,*,debugNone):ifevents._get_running_loop()isnotNone:raiseRuntimeError(asyncio.run() cannot be called from a running event loop)loopevents.new_event_loop()try:events.set_event_loop(loop)ifdebugisnotNone:loop.set_debug(debug)returnloop.run_until_complete(main)finally:try:_cancel_all_tasks(loop)loop.run_until_complete(loop.shutdown_asyncgens())finally:events.set_event_loop(None)loop.close()注意第一行检查events._get_running_loop() is not None。如果当前线程里已经有一个事件循环在跑了asyncio.run()会直接抛出RuntimeError拒绝创建第二个循环。这个设计是合理的——它防止了在同一个线程里出现两个事件循环竞争资源的情况。但问题出在调用链的隐蔽性上你写的代码没有直接调asyncio.run()但你依赖的某个第三方库内部调了。或者你在一个已经运行在事件循环里的上下文比如 FastAPI 路由、Jupyter cell里又写了一个asyncio.run()。典型触发场景一FastAPI 异步推理脚本FastAPI 运行在 uvicorn 之上uvicorn 启动时就会创建并运行一个事件循环。所有async def路由函数都在这个循环上调度。如果你的推理脚本内部封装了asyncio.run()嵌套就发生了# main.pyfromfastapiimportFastAPIfrominferenceimportrun_inference appFastAPI()app.get(/predict)asyncdefpredict(input:str):# 此时已经运行在 uvicorn 的事件循环中resultrun_inference(input)# ← 内部调了 asyncio.run()return{result:result}# inference.pyimportasyncioasyncdef_model_infer(input_data):awaitasyncio.sleep(1)# 模拟异步推理returnfprocessed:{input_data}defrun_inference(input_data):# 致命在已有事件循环的上下文中再创建新循环returnasyncio.run(_model_infer(input_data))启动服务并请求$ uvicorn main:app--reload$curlhttp://localhost:8000/predict?inputtest报错RuntimeError: asyncio.run() cannot be called from a running event loop典型触发场景二Jupyter NotebookIPython 7.0 之后notebook 的 kernel 默认运行在 asyncio 事件循环中。这意味着你在 cell 里直接写asyncio.run(some_coroutine())也会触发同样的错误。# Jupyter Cellimportasyncioasyncdefhello():awaitasyncio.sleep(1)returnhelloasyncio.run(hello())# ← 报错event loop is already running三、解法一nest_asyncio 补丁nest_asyncio是一个第三方库作者 Ewald de Wit它通过猴子补丁monkey patch修改了 asyncio 内部_get_running_loop()和run_until_complete()的行为使得同一个线程内可以嵌套运行事件循环。importnest_asyncio nest_asyncio.apply()# 一次性全局打补丁importasyncioasyncdefinner():awaitasyncio.sleep(1)returninner doneasyncdefouter():# 正常情况下这里会报错# 打了补丁后asyncio.run() 会复用当前循环resultasyncio.run(inner())returnfouter got:{result}# 在 Jupyter 中可以直接 await不需要 asyncio.run()# await outer()# 在普通脚本中即使嵌套调用也能正常运行asyncio.run(outer())它到底改了什么nest_asyncio的核心修改是把BaseEventLoop.run_until_complete()替换为一个允许递归调用的版本。原版在调用时会检查_running标志如果已经是 running 状态就报错补丁版会暂时把当前循环的_running设为 False让内层的run_until_complete()顺利通过检查执行完毕后再恢复状态。# nest_asyncio 的核心逻辑极度简化def_run_until_complete(self,future):# 原版的 _run_until_complete 会检查 self._running# nest_asyncio 把它改成如果已经在 running就递归调度ifself._running:# 把当前 future 放入队列等待当前循环的下一次迭代执行returnself._run_once_or_until(future)# 否则走原版逻辑...适用边界场景是否推荐Jupyter Notebook 调试推荐这是 nest_asyncio 最常见的用法本地快速验证第三方库可以接受生产环境的 FastAPI 服务不推荐。它改变了 asyncio 的标准语义在复杂并发场景下大量 Task 同时创建、取消、异常处理可能引入不可预期的行为四、解法二异步任务拆分重构最干净的做法问题产生的根本原因是异步函数的内部又包了一层同步入口而这个同步入口擅自创建了事件循环。把同步入口去掉让异步逻辑自然地向调用链上层暴露就能彻底解决这个问题。重构前的错误结构# inference.py —— 问题版本asyncdef_model_infer(input_data):resultawaitsome_async_api(input_data)returnresultdefrun_inference(input_data):# ← 同步包装层是罪魁祸首returnasyncio.run(_model_infer(input_data))重构后的正确结构# inference.py —— 正确版本asyncdefmodel_infer(input_data):# ← 直接暴露 async 接口resultawaitsome_async_api(input_data)returnresult# 删除 run_inference() 这个同步包装层FastAPI 路由层自然调用# main.pyfromfastapiimportFastAPIfrominferenceimportmodel_infer# 直接导入 async 函数appFastAPI()app.post(/predict)asyncdefpredict(input:str):resultawaitmodel_infer(input)# ← 复用外层事件循环return{result:result}这个重构的实质是让事件循环的管理权上移到调用链的最顶层uvicorn/FastAPI所有下游的异步函数只负责业务逻辑不再自己起炉灶。如果调用方是同步代码比如一个命令行脚本怎么办# cli.py —— 同步入口importasynciofrominferenceimportmodel_inferdefmain():resultasyncio.run(model_infer(test))# 唯一的事件循环在这里创建print(result)if__name____main__:main()这样事件循环的生命周期是清晰的只在main()里创建一次所有异步函数都挂载到这一个循环上。五、解法三子线程独立事件循环如果第三方库的源码无法修改且它内部固执地调了asyncio.run()那就只能把冲突隔离到不同的线程里。每个线程可以拥有独立的事件循环它们互不干扰。importasyncioimportconcurrent.futuresfromfastapiimportFastAPI appFastAPI()# 线程池在应用级别创建生命周期与进程一致_thread_poolconcurrent.futures.ThreadPoolExecutor(max_workers4)asyncdef_third_party_infer_impl(input_data):# 第三方库内部的异步实现awaitasyncio.sleep(0.5)returnfinferred:{input_data}defthird_party_infer(input_data):# 模拟不可修改的第三方库同步接口内部自己调 asyncio.run()returnasyncio.run(_third_party_infer_impl(input_data))app.get(/predict)asyncdefpredict(input:str):loopasyncio.get_running_loop()# 把同步函数提交到线程池resultawaitloop.run_in_executor(_thread_pool,third_party_infer,input)return{result:result}run_in_executor的工作机制loop.run_in_executor(executor, func, *args)会把func提交到一个线程池默认是ThreadPoolExecutor中执行同时返回一个asyncio.Future。await这个 Future 时当前协程会挂起等待线程池中的func执行完毕。func的执行不会阻塞事件循环——它是在另一个线程里跑的。当func内部调了asyncio.run()时由于它运行在新线程中该线程原先没有事件循环所以asyncio.run()可以正常创建并运行一个新的循环。主线程的事件循环和子线程的事件循环完全隔离不存在嵌套冲突。需要注意的坑线程安全如果_run_in_new_loop内部访问了共享状态比如全局变量、数据库连接需要加锁。asyncio的锁不能跨线程使用需要用threading.Lock()上下文变量ContextVarsasyncio的contextvars不会自动传播到子线程。如果依赖上下文变量传递请求 ID、用户信息等需要手动传递性能开销线程切换和 GIL 竞争会带来额外开销。如果推理任务本身是纯异步 IO 的没有 CPU 密集计算放到线程里反而多绕了一层六、三种解法的对比与决策维度nest_asyncio异步重构子线程隔离改动成本极低2 行代码中等需重构接口较高需处理线程安全代码侵入性无侵入但改变标准库行为需修改调用链接口需包装层 线程池管理副作用风险有改变 asyncio 语义无有线程安全、上下文传递长期维护性差好中等适用场景调试、原型验证自己可控的代码库不可修改的第三方库决策路径能否修改源码 ├─ 是 → 选择异步重构最佳实践 └─ 否 → 是否为生产环境 ├─ 是 → 选择子线程隔离 └─ 否 → 选择nest_asyncio快速验证七、工程原则事件循环的生命周期管理最后归纳几条实践中总结的原则避免反复踩坑原则一一个应用只有一个事件循环入口无论是 uvicorn、Celery worker还是命令行脚本事件循环的创建和关闭应该只发生在一个地方。不要让业务代码自己去asyncio.run()或loop.run_until_complete()。原则二async def函数不创建事件循环# 错误asyncdefprocess():resultasyncio.run(some_other_async())# 自己创建了循环# 正确asyncdefprocess():resultawaitsome_other_async()# 交给调用方的循环调度原则三同步函数如果需要异步结果使用asyncio.run()作为最外层入口# 正确同步入口调用异步逻辑defmain():resultasyncio.run(async_main())# 错误同步函数内部层层嵌套 asyncio.run()defhelper():returnasyncio.run(async_helper())# 如果 helper 被另一个 async 函数调用就炸了原则四区分 “async 函数” 和 “返回 async 结果的同步函数”有些库提供的接口看起来是同步的但内部偷偷调了asyncio.run()。接入这类库时务必读一下它的源码或文档确认它的事件循环管理策略。如果它自己管循环就不要在已有循环的上下文里调用它要么改它的源码重构为 async 接口要么用子线程隔离。八、总结RuntimeError: This event loop is already running不是一个随机的报错它是 asyncio 设计哲学的外在体现一个线程、一个循环、统一调度。理解了事件循环的单线程约束就能明白为什么asyncio.run()要有那道防御性检查以及三种解法各自解决的是哪一层的问题。nest_asyncio绕过检查适合调试不适合生产异步重构把事件循环的管理权上移到顶层最干净最推荐子线程隔离用线程边界隔离冲突适用于不可控的第三方库事件循环的管理本质上是一种资源所有权问题——谁创建、谁调度、谁关闭必须在设计之初就确定下来。如果调用链上的每一层都想自己起一个新循环最终必然撞墙。