Python asyncio.Runner 怎样复用同一个事件循环运行多个入口

我在把几个异步入口塞进同一个命令行宿主时，最先遇到的误区是连续调用 asyncio.run()。它确实能让每个协程跑完，但每次调用都会负责创建、运行并关闭自己的事件循环；如果几个入口需要共享同一个循环、同一份 contextvars.Context 或同一套运行时配置，asyncio.Runner 更合适。

官方地址：https://docs.python.org/3/library/asyncio-runner.html

需要一次性执行一个顶层异步入口时继续用 asyncio.run()；需要在同一个同步宿主里多次调用顶层异步函数时，用 with asyncio.Runner() as runner 包住调用范围，并把关闭责任交给上下文管理器。

先把一次性入口和复用型运行时分开

asyncio.run(coro) 的定位是程序入口：它创建事件循环，运行可等待对象，收尾异步生成器，关闭默认执行器，然后关闭事件循环。这样的边界很干净，短脚本、单次命令和简单测试都不需要额外管理运行时。

asyncio.Runner 则是一个上下文管理器。它采用延迟初始化策略，在进入 with 代码块、第一次调用 run() 或调用 get_loop() 时建立嵌入式事件循环与上下文。多个 runner.run() 调用可以落在同一个 Runner 作用域内，适合批处理器、测试夹具和需要连续执行多个异步入口的同步宿主。

用 Runner 让多个入口共享同一循环

最小写法不是把事件循环对象到处传给业务函数，而是让同步宿主持有 Runner，把每个顶层异步入口作为一次独立的 run() 调用。这样业务协程仍然只接收自己的参数，运行时生命周期集中在宿主层。

import asyncio


async def load_catalog():
    # 代表第一段异步工作，实际项目中可以替换成数据库或网络读取。
    await asyncio.sleep(0)
    return ["book", "movie"]


async def build_report(items):
    # 第二个入口只依赖显式参数，不自行创建或关闭事件循环。
    await asyncio.sleep(0)
    return {"count": len(items), "items": items}


def run_batch():
    # Runner 的作用域覆盖多个顶层入口，共享同一个嵌入式运行时。
    with asyncio.Runner() as runner:
        items = runner.run(load_catalog())
        report = runner.run(build_report(items))
        return report


if __name__ == "__main__":
    # 同步入口只负责选择运行时，不把循环管理泄漏给业务函数。
    print(run_batch())

这里的关键不是把两个协程“排成一条流水线”，而是让它们处于同一个 Runner 的生命周期边界内。调用之间仍然要通过返回值或显式上下文传递数据；不要依赖某个已经关闭的循环，也不要把 runner.get_loop() 当成全局单例。

把 loop_factory 和 Context 放到装配层

当项目需要统一配置事件循环时，Runner 的 loop_factory 是比业务协程内部调用 new_event_loop() 更清晰的装配点。官方文档说明，自定义工厂需要负责创建并设置当前事件循环；不传工厂时，Runner 使用默认的新事件循环并设置为当前循环。

import asyncio
import contextvars


request_id = contextvars.ContextVar("request_id", default="unknown")


def make_loop():
    # 工厂只负责创建并登记循环，调试和资源策略集中在这里配置。
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    return loop


async def show_request():
    # 当前入口从 Runner 的上下文读取请求标识，不改动全局变量。
    return request_id.get()


def run_with_context(value):
    # 每次入口调用都可以显式传入自己的 Context，避免上下文串线。
    context = contextvars.copy_context()
    context.run(request_id.set, value)
    with asyncio.Runner(loop_factory=make_loop) as runner:
        return runner.run(show_request(), context=context)


print(run_with_context("job-17"))

context 参数适合需要显式隔离请求标识、租户信息或追踪字段的场景。它和事件循环不是同一个概念：循环负责调度，Context 负责一组上下文变量的取值。把两者放在宿主装配层，业务函数就不需要知道自己是由 asyncio.run() 还是 Runner 启动的。

关闭责任和嵌套限制要提前定清

Runner.close() 会最终化异步生成器、关闭默认执行器、关闭事件循环，并释放嵌入的上下文。使用 with asyncio.Runner() as runner 可以让这些动作在作用域退出时统一发生；如果手动创建 Runner，就必须明确谁负责调用 close()，否则长生命周期宿主可能积累资源。

另一个容易踩到的边界是：runner.run() 不能在同一线程已经运行事件循环时调用。这和 asyncio.run() 不能嵌套调用的限制属于同一类问题。Web 框架、Notebook 或其他异步宿主已经持有运行中的循环时，应该直接把工作写成 await，而不是再包一层 Runner。

import asyncio


async def consume():
    # finally 用于释放业务资源；不要把取消异常当普通成功吞掉。
    try:
        await asyncio.sleep(0)
        return "done"
    finally:
        # 这里放连接、文件或临时句柄的释放逻辑。
        pass


def main():
    # 上下文管理器退出时负责 Runner 的统一关闭。
    with asyncio.Runner() as runner:
        return runner.run(consume())


if __name__ == "__main__":
    print(main())

如果某个入口抛出异常，异常会从 runner.run() 返回给同步宿主；是否重试、跳过后续入口或终止批次，应由宿主层作出决定。不要在每个入口里隐式关闭共享资源，也不要为了“让下一项继续”而把所有异常都改成成功结果。

按负载选择入口方案

我自己的判断是：如果调用方只需要“把一个协程跑完”，asyncio.run() 的简单性更有价值；如果调用方需要把多个顶层入口放在同一个同步控制面里，Runner 才能把运行时边界表达清楚。不要为了“复用”而把所有脚本都改成 Runner，少一个生命周期对象本身也是一种工程收益。

常见问题

Runner 会让多个入口自动并发执行吗？

不会。多个 runner.run() 调用共享同一个运行时，但每次调用仍由同步宿主决定何时开始和如何传递结果。需要并发时，应在某个顶层协程内部使用任务组或其他并发原语。

可以把 Runner 做成全局变量吗？

通常不建议。全局 Runner 会模糊关闭责任，并让测试之间共享上下文和事件循环。更稳妥的方式是让命令、批次或测试夹具持有一个明确作用域的 Runner。

loop_factory 一定要自己调用 set_event_loop 吗？

自定义工厂应负责把创建的循环设置为当前循环；官方文档把这项责任交给工厂。若不需要自定义创建逻辑，直接省略 loop_factory，使用 Runner 的默认行为更简单。

为什么在 Jupyter 里调用 Runner 会报循环已运行？

Notebook 通常已经在当前线程运行事件循环，而 Runner 的 run() 不能在同一线程嵌套运行。把代码改成当前单元格直接 await，或把异步工作交给 Notebook 提供的宿主机制。

asyncio.Runner 的价值不是替代 asyncio.run()，而是把“多个顶层入口共享一个异步运行时”这个需求表达出来。先判断调用方是否已经拥有事件循环，再决定使用一次性入口、Runner 作用域还是直接 await；运行时和业务代码分层，关闭责任也会更容易维护。