CPython asyncio 运行器（Runners）权威指南：asyncio.run 与 Runner 的用法、源码原理与 Ctrl-C 中断处理

2026-09-06 18:52:28作者：管翌锬

导读

本文以 CPython 官方文档 Doc/library/asyncio-runner.rst 为主线，系统讲解 asyncio 的两大高层运行原语：asyncio.run() 与 asyncio.Runner 上下文管理器。你将掌握它们的事件循环/执行器生命周期管理机制、debug 与 loop_factory 参数的语义、多顶层协程共享同一事件循环与 contextvars 上下文的实战写法，以及 asyncio 如何通过自定义 SIGINT 处理器优雅地处理 Ctrl-C 中断（Python 3.11+）。文中结合 Lib/asyncio/runners.py 源码与其单元测试 Lib/test/test_asyncio/test_runners.py，从实现层面印证每个行为细节。

一、asyncio.run()：asyncio 程序的统一入口

1.1 函数签名与语义

asyncio.run() 是最常用的高层入口，用于在事件循环中执行一个 awaitable 并返回其结果：

asyncio.run(coro, *, debug=None, loop_factory=None)

参数与行为要点（依源码 Lib/asyncio/runners.py 印证）：

coro：自 Python 3.14 起可以是任意 awaitable 对象（协程、asyncio.Future、实现了 __await__ 的对象等）。若传入的是协程则直接执行；若是其他 awaitable，源码中会将其包装进一个内部协程再运行（见 runners.py）。如果传入的参数既不是协程也不是 awaitable，会抛出 TypeError: An asyncio.Future, a coroutine or an awaitable is required。
返回值：coro 的执行结果；若协程内部抛出异常，则该异常会原样向上抛出。单元测试 test_asyncio_run_return/test_asyncio_run_raises 分别验证了返回结果与异常传播（见 test_runners.py）。

该函数为「一站式」封装：它负责管理事件循环的创建与关闭、终结异步生成器（async generators）、关闭默认执行器（default executor），并保证在函数退出前完成这些清理动作。

1.2 debug 参数：三态语义

debug 是一个三态参数：

取值	行为
`True`	显式以调试模式运行事件循环
`False`	显式禁用调试模式
`None`（默认）	遵循全局调试模式设置

其中 None（自 Python 3.10 起成为默认值）意味着不强制覆盖全局配置。全局的 asyncio 调试模式可以通过以下方式开启（详见 Doc/library/asyncio-dev.rst）：

设置环境变量 PYTHONASYNCIODEBUG=1；
使用 Python Development Mode（-X dev / PYTHONDEVMODE=1）；
显式传入 debug=True（等价于调用 loop.set_debug(True)）。

从源码看，Runner.__init__ 只是保存 debug 值，真正的生效发生在 _lazy_init() 阶段：只有 debug is not None 时才会调用 loop.set_debug(self._debug)（见 runners.py）。这正解释了「None 尊重全局设置」的实现机理——不主动调用 set_debug，调试开关便由事件循环创建时的全局默认决定。测试 test_asyncio_run_debug 用 mock 模拟了全局调试开关的开/关两种状态，验证了 debug=None 与显式传值之间的覆盖关系（见 test_runners.py）。

1.3 loop_factory 参数：自定义事件循环创建

自 Python 3.12 起新增 loop_factory 参数。它用于接管事件循环的创建逻辑：

若 loop_factory is None，使用 asyncio.new_event_loop() 创建循环；
否则调用 loop_factory() 的返回值作为事件循环，并在结束后将其关闭。

官方文档明确建议：需要通过自定义方式配置事件循环时，优先使用 loop_factory。典型应用包括：

替换为第三方事件循环实现（如基于 libuv 的 uvloop 风格循环）；
注入带自定义 Task 工厂或自定义默认执行器配置的事件循环；
在测试中注入可 mock 的循环对象（测试套件大量使用该手法）。

测试 test_asyncio_run_loop_factory 验证了 factory 恰好被无参调用一次，且 asyncio.get_running_loop() 返回的正是该工厂创建的循环（见 test_runners.py）；test_loop_factory_default_event_loop 则验证了在 Windows 平台默认得到 ProactorEventLoop、其他平台默认得到 SelectorEventLoop（test_runners.py）。

一个可直接运行的最小示例：

import asyncio

async def main():
    await asyncio.sleep(1)
    print('hello')

asyncio.run(main())

说明：本仓库文档面向 Python 3.14 开发版本，示例与上述特性均以当前仓库代码为准。

1.4 底层调用链：run() 不过是 Runner 的薄封装

查看 runners.py 可以发现 asyncio.run() 的全部实现只有两步：

def run(main, *, debug=None, loop_factory=None):
    if events._get_running_loop() is not None:
        # fail fast with short traceback
        raise RuntimeError(
            "asyncio.run() cannot be called from a running event loop")
    with Runner(debug=debug, loop_factory=loop_factory) as runner:
        return runner.run(main)

即：run() = 「检查当前线程没有正在运行的循环」+「进入 Runner 上下文并调用一次 runner.run()」。因此它天然具备 Runner 的全部清理能力：

进入 with 时通过 _lazy_init() 惰性创建事件循环并设为当前循环；
runner.run(main) 将协程包装为 Task 执行；
退出 with 时调用 Runner.close()，依次完成取消遗留任务 → 关闭异步生成器 → 关闭默认执行器 → 解除当前事件循环设置 → 关闭循环。

asyncio.run() 因此强调「每个程序只在顶层调用一次」，对应其一次性创建并最终关闭循环的语义；run() 也不能在线程中已有正在运行的事件循环时再次调用（见 runners.py），否则抛出 RuntimeError，测试 test_asyncio_run_from_running_loop 覆盖了该场景（test_runners.py）。

1.5 默认执行器的 5 分钟关闭超时

当 run()/Runner.close() 关闭默认执行器（线程池）时，会给执行器 5 分钟的退出时间：

若执行器内的线程在时限内正常结束，则平滑关闭；
若超时未完成，则发出警告并强制关闭执行器。

该超时值定义在常量 THREAD_JOIN_TIMEOUT = 300（秒）中，见 Lib/asyncio/constants.py；实际由 loop.shutdown_default_executor(timeout) 负责执行（实现见 Lib/asyncio/base_events.py，自 Python 3.9 起 run() 改为调用该方法完成清理）。

1.6 版本演进一览

版本	变更内容
3.7	新增 `asyncio.run()`
3.9	改用 `loop.shutdown_default_executor` 完成执行器清理
3.10	`debug` 默认值改为 `None`，尊重全局调试模式设置
3.12	新增 `loop_factory` 参数
3.14	`coro` 参数放宽为任意 awaitable 对象

二、Runner 上下文管理器：同上下文多次运行顶层协程

2.1 设计动机与适用场景

asyncio.Runner（Python 3.11 新增）是一个上下文管理器，用于在同一个事件循环和同一个 contextvars.Context 上下文中，多次调用多个顶层异步函数。

这在以下场景中非常关键（源码注释见 runners.py）：

交互式控制台 / IPython：用户逐条执行异步语句，需要复用同一个循环与上下文；
unittest 运行器：多个测试用例需要共享事件循环；
命令行工具：从一个同步主框架中反复调用异步代码——凡是「多次调用应保持上下文连续」且单个 asyncio.run() 不适用的地方，Runner 都是首选。

asyncio.run() 可以看作是 Runner 在单次调用场景下的便捷缩写（源码 docstring 原文），二者关系如下：

# asyncio.run(main(), debug=True) 等价于：
with asyncio.Runner(debug=True) as runner:
    runner.run(main())

官方文档给出的完整改写示例：

import asyncio

async def main():
    await asyncio.sleep(1)
    print('hello')

with asyncio.Runner() as runner:
    runner.run(main())

2.2 构造参数与生命周期

asyncio.Runner(*, debug=None, loop_factory=None)

debug：与 run() 相同的三态语义（True/False/None 遵循全局设置），由 _lazy_init 在创建循环后立即应用；
loop_factory：覆盖事件循环的创建方式。关键约束是：自定义 loop_factory 有责任把创建的循环设为当前事件循环（因为只有走默认 new_event_loop() 路径时 Runner 才会自动调用 set_event_loop）。当 loop_factory is None 时，Runner 使用 new_event_loop() 创建循环，并通过 set_event_loop() 将其设为当前循环。

Runner 在其文档字符串中标注为 final 类，不打算被继承（见 runners.py）。

生命周期方面，Runner 采用惰性初始化策略：

构造函数（runners.py）只保存参数并初始化状态字段，不创建任何底层结构；
内嵌的事件循环与 contextvars.Context 在以下三个时机之一才真正创建：进入 with 语句体时、第一次调用 run() 时、第一次调用 get_loop() 时。

该判断在 _lazy_init() 中集中实现（runners.py）：

def _lazy_init(self):
    if self._state is _State.CLOSED:
        raise RuntimeError("Runner is closed")
    if self._state is _State.INITIALIZED:
        return
    if self._loop_factory is None:
        self._loop = events.new_event_loop()
        if not self._set_event_loop:
            # 只调用一次 set_event_loop，避免对子 watcher 重复 attach_loop
            events.set_event_loop(self._loop)
            self._set_event_loop = True
    else:
        self._loop = self._loop_factory()
    if self._debug is not None:
        self._loop.set_debug(self._debug)
    self._context = contextvars.copy_context()
    self._state = _State.INITIALIZED

值得注意的源码细节：

set_event_loop 通过 _set_event_loop 标志保证最多只调用一次，即使 run() 被调用多次也不会重复设置（回归测试见 test_set_event_loop_called_once，test_runners.py）；注释说明这样做是为了避免对子进程 watcher 重复调用 attach_loop。
Runner 内部用 contextvars.copy_context() 快照创建时的上下文，保证所有在 Runner 中运行的协程共享该上下文基线。

Runner 内部有一个三态状态机（runners.py）：

class _State(enum.Enum):
    CREATED = "created"      # 已构造，尚未初始化底层资源
    INITIALIZED = "initialized"  # 循环与上下文已就绪
    CLOSED = "closed"        # 已关闭，任何后续操作都会报错

2.3 Runner.run()：带自定义上下文执行协程

Runner.run(coro, *, context=None)

在 Runner 内嵌的事件循环中执行 coro（Python 3.14 起可为任意 awaitable，实现上通过 _wrap_awaitable 包装非协程的 awaitable）；
若传入的是协程，则会将其包装成一个 Task（self._loop.create_task(coro, context=context)，见 runners.py）；
可选关键字参数 context 允许指定一个自定义的 contextvars.Context 供这段代码运行；当 context=None 时使用 Runner 的默认上下文（即进入时快照的那份）；
返回 awaitable 的执行结果，或原样抛出异常；
同样不能在线程中已有正在运行的事件循环时被调用，会抛出 RuntimeError: Runner.run() cannot be called from a running event loop。

方法可以在同一 with 块内反复调用多次，这是它与 asyncio.run() 最大的区别。测试 test_run_keeps_context 生动展示了「上下文跨多次调用延续」的价值（test_runners.py）：

cvar = contextvars.ContextVar("cvar", default=-1)

async def f(val):
    old = cvar.get()
    await asyncio.sleep(0)
    cvar.set(val)
    return old

with asyncio.Runner() as runner:
    assert -1 == runner.run(f(1))   # 读到默认值 -1，随后把 cvar 设为 1
    assert 1 == runner.run(f(2))    # 读到上次设置的 1，随后把 cvar 设为 2
    assert 2 == runner.run(copy_current_context()).get(cvar)  # 上下文确实被延续

2.4 Runner.close() 与 Runner.get_loop()

close()：关闭 Runner，完成四步清理（见 runners.py）：

_cancel_all_tasks(loop)——取消当前循环内所有未完成任务，并用 gather(..., return_exceptions=True) 等待它们结束；若某任务在取消后又抛出了其他异常，会通过 loop.call_exception_handler 上报 unhandled exception during asyncio.run() shutdown（见 runners.py）；
loop.shutdown_asyncgens()——终结所有异步生成器；
loop.shutdown_default_executor(constants.THREAD_JOIN_TIMEOUT)——以 300 秒超时关闭默认执行器；
若 Runner 曾经设置过当前事件循环（_set_event_loop 为真），则调用 set_event_loop(None) 解除绑定；最后 loop.close() 关闭循环并置 _state = CLOSED。

get_loop()：返回 Runner 内嵌的事件循环对象，会在首次调用时触发惰性初始化。关闭后再调用 get_loop() 会抛出 RuntimeError: Runner is closed。

单元测试还验证了若干边界行为：

Runner 退出 with 后循环确实被关闭（test_run，test_runners.py）；
显式调用 close() 与重复 close() 均是幂等安全的（test_explicit_close、test_double_close）；
Runner 退出 with 后不能再次进入（test_second_with_block_raises，test_runners.py）。

一个包含「遗留后台任务」的综合示例，用于演示 Runner 退出时对悬挂任务的处理（对应 test_asyncio_run_cancels_hanging_tasks 的行为，test_runners.py）：

import asyncio

async def leftover():
    await asyncio.sleep(0.1)

async def main():
    lo_task = asyncio.create_task(leftover())  # main 返回后它仍悬挂
    return 123

with asyncio.Runner() as runner:
    result = runner.run(main())   # result == 123
# 退出 with 时，leftover 会被 Runner 自动取消并等待完成

三、KeyboardInterrupt / Ctrl-C 的可靠处理

本节特性自 Python 3.11 起生效。

3.1 背景：为什么默认的 Ctrl-C 处理对 asyncio 不适用

当用户按 Ctrl-C 触发 SIGINT 信号时，Python 默认在主线程抛出 KeyboardInterrupt。然而这套机制与 asyncio 并不兼容：KeyboardInterrupt 可能打断 asyncio 内部的任意代码路径（如事件循环的选择器轮询、Future 回调分发等），造成资源未释放甚至程序挂起无法退出。

3.2 asyncio 的四步缓解机制

为缓解该问题，asyncio.Runner.run() 对 SIGINT 做了专门处理（文档原文，配合 runners.py 与 runners.py 的 _on_sigint 实现）：

第 1 步：安装自定义 SIGINT 处理器。 Runner.run() 在执行任何用户代码之前，先检查当前线程是否为主线程、当前 SIGINT 处理器是否仍是默认处理器（signal.default_int_handler）；满足条件则用 functools.partial 绑定主任务安装自定义处理器，并在函数退出时移除、恢复默认处理器。若当前线程不支持信号注册（例如信号未启用的嵌入式解释器场景，见 gh-91880 注释），signal.signal 会抛出 ValueError，Runner 会将其吞掉并跳过安装。

第 2 步：创建主任务。 Runner 为传入的协程创建主 Task（self._loop.create_task(...)），后续中断操作都作用在该任务上。

第 3 步：取消主任务并重抛 KeyboardInterrupt。 当 Ctrl-C 触发自定义处理器 _on_sigint 时：

第一次触发（_interrupt_count == 1）且主任务尚未完成，则调用 task.cancel()，在主任务内部抛出 CancelledError——这会引发 Python 栈的正常展开，使用户代码中的 try/except、try/finally 得以执行资源清理；
同时通过 call_soon_threadsafe(lambda: None) 唤醒可能阻塞在 select() 长超时中的事件循环；
主任务被取消后，Runner.run() 会抛出一个 KeyboardInterrupt 给调用方。

第 4 步：双 Ctrl-C 强制中断。 若用户代码是 Task.cancel() 无法中断的紧密循环（例如持续让出却不检查取消），第二次按 Ctrl-C 会立即直接抛出 KeyboardInterrupt，不再执行取消流程，保证程序始终能被用户强制终止。

从实现细节看，CancelledError → KeyboardInterrupt 的转换发生在 run() 的异常处理分支（runners.py）：当捕获到 CancelledError 且 _interrupt_count > 0 时，通过 task.uncancel() 检查取消计数是否归零，若归零则升级为 KeyboardInterrupt 抛出。为兼容没有 uncancel 方法的旧 Task 实现，这里使用了 getattr(task, "uncancel", None) 防御式取值（相关回归见测试 test_asyncio_run_without_uncancel，test_runners.py）。

3.3 中断行为的测试佐证

Lib/test/test_asyncio/test_runners.py 中专门有一组 test_interrupt_* 用例验证上述机制，可用 python -m unittest test.test_asyncio.test_runners 运行：

test_interrupt_call_soon：任务不等待任何 Future 而是紧循环时，第一次 Ctrl-C 取消任务，最终抛出 KeyboardInterrupt；
test_interrupt_wait：等待 Future 时被中断，会同时取消该 Future 与主任务；
test_interrupt_cancelled_task：若主任务已被主动取消，中断不会误抛 KeyboardInterrupt，而是保持 CancelledError 语义；
test_signal_install_not_supported_ok：主线程不支持信号注册时，Runner 仍能正常运行。

四、总结与选型建议

将本文内容落实到实际编码选择上：

绝大多数单次运行的脚本（脚本的主入口、__main__ 块）：直接使用 asyncio.run(coro)，让函数自动完成循环创建、协程执行与资源清理。
需要自定义循环实现或深度配置循环：使用 asyncio.run(coro, loop_factory=my_factory) 或 asyncio.Runner(loop_factory=my_factory)，注意自定义工厂必须自行负责将循环设为当前循环。
需要多次运行顶层协程并共享上下文（交互控制台、测试框架、同步框架内嵌 async）：使用 with asyncio.Runner() as runner:，并在块内多次调用 runner.run(coro)；需要隔离上下文时可传入 context= 参数。
希望获得健壮的 Ctrl-C 行为：只要通过 Runner.run()（包括 asyncio.run()）进入，即可自动获得「先取消任务→资源清理→最终 KeyboardInterrupt」的优雅中断保证，以及第二次 Ctrl-C 强制退出的兜底。