Python异步编程中怎么正确处理异常与超时

在 Python 的异步编程（asyncio）中，正确处理异常与超时不仅是让代码在报错时不崩溃，更关乎底层协程任务的生命周期管理、事件循环资源的回收以及取消信号的精准传递。处理不当容易引发任务静默失败、资源泄露或死锁。理解异常如何沿 await 调用链传播、掌握新旧超时控制 API 的本质区别，并配合恰当的并发容器管理兄弟任务，是构建高可用异步系统的关键路径。

一、协程内部异常的传播与捕获机制

在单协程的同步调用逻辑中，捕获异常与传统 Python 代码十分类似。当一个被 await 的协程抛出异常时，该异常会立即沿着调用栈向上冒泡，并在当前协程的 await 表达式位置被抛出。只要在此处包裹标准的 try...except 语句，就能像普通同步函数一样捕获并处理错误。

然而，初学者最常踩坑的场景在于通过 asyncio.create_task() 启动后台并发任务。如果创建的任务在运行期间发生异常，且主流程没有主动 await 该任务或没有为其添加回调，该异常不会立即打断主事件循环，而是被缓存在任务对象内部。直到该任务被垃圾回收时，事件循环才会输出一条警告日志（Task exception was never retrieved）。这意味着业务逻辑可能已经失败，但调用方完全没有感知。

import asyncio

async def faulty_fetch(url: str):
    await asyncio.sleep(0.1)
    if "error" in url:
        raise ValueError(f"请求目标异常: {url}")
    return f"内容来自 {url}"

async def main():
    # 做法一：直接 await 协程，异常在原地抛出并捕获
    try:
        result = await faulty_fetch("https://example.com/error")
    except ValueError as err:
        print(f"成功捕获直接异常: {err}")

    # 做法二：创建 Task 但未妥善保护可能遗漏异常
    task = asyncio.create_task(faulty_fetch("https://example.com/error"))
    # 必须显式 await task 才能在调用点暴露该任务内部积累的异常
    try:
        await task
    except ValueError as err:
        print(f"成功从 Task 实例捕获异常: {err}")

asyncio.run(main())

二、超时控制的两种核心范式与关键差异

在异步网络请求或重计算中，为操作设定超时上限是保证系统响应性的底线。Python 提供了两种主要方式来实现超时：早期引入的 asyncio.wait_for() 函数，以及 Python 3.11 加入的 asyncio.timeout() 上下文管理器。深入理解这两者的差异对于编写现代异步代码至关重要。

1. 传统方式：asyncio.wait_for()

asyncio.wait_for(aw, timeout) 接收一个可等待对象并设置最长耗时（秒）。一旦超时触发，它会主动调用该底层的 task.cancel()，随后抛出 asyncio.TimeoutError（在 3.11 以前为特定别名，3.11 起统一为内置的 TimeoutError）。

import asyncio

async def slow_operation():
    try:
        await asyncio.sleep(5)
        return "完成"
    except asyncio.CancelledError:
        print("slow_operation 被取消，执行收尾工作")
        raise  # 必须重新向上抛出取消异常

async def test_wait_for():
    try:
        await asyncio.wait_for(slow_operation(), timeout=1.0)
    except TimeoutError:
        print("操作已超时，底层协程已收到取消信号并退出")

2. 现代推荐：asyncio.timeout() 上下文管理器

Python 3.11 引入了 async with asyncio.timeout(delay):。它直接作用于代码块级别，而非单个函数调用。它在语义、性能以及多重嵌套下的取消信号识别上具备显著优势。

import asyncio

async def test_timeout_ctx():
    try:
        async with asyncio.timeout(1.5):
            # 可以在同一个超时配额下顺序执行多个 await 步骤
            await asyncio.sleep(0.5)
            await asyncio.sleep(0.5)
            await asyncio.sleep(1.0)  # 此处总时长达到 2.0s，触发超时
    except TimeoutError:
        print("整个上下文代码块超时，未完成的任务自动被取消")

关键差异为什么重要：

• 范围与粒度：asyncio.wait_for 每次只能包裹单一可等待对象；若要对“建立连接 + 发送请求 + 读取响应”一连串操作设置整体超时上限，必须将它们包装进一个新协程函数。而 asyncio.timeout 是上下文管理器，可直接包裹整个执行块，所有内部顺序执行的子步骤共享相同的截止时间。
• 取消归因准确性：当代码块本身正在被外部其他协程取消时，asyncio.wait_for 有时容易将内部任务抛出的 CancelledError 误认为是自己超时产生的取消，导致把普通的取消混淆为 TimeoutError。asyncio.timeout 内部维护了取消计数器与独立令牌，能够精确判断当前取消究竟是由自身超时触发，还是来自外层的父任务取消。

三、并发场景下的异常隔离：TaskGroup 与 gather

在实际生产环境中，绝大多数异步程序都会并发执行多个子任务。此时处理异常面临两种截然不同的诉求：要么是“只要有一个任务失败，其余任务全部立刻取消并向上报告”；要么是“允许部分任务失败，汇总所有结果和异常逐个分析”。

1. 结构化并发标准：asyncio.TaskGroup（推荐）

Python 3.11 引入了结构化并发的核心组件 asyncio.TaskGroup。在 async with asyncio.TaskGroup() as tg: 块中创建的所有任务，当其中任何一个任务抛出非取消异常时，TaskGroup 会立即向该组内的其他所有子任务发送取消信号，并等待它们全部退出。随后，所有未被内部捕获的异常会被打包成一个 ExceptionGroup 抛出，配合 except* 语法可以实现精准分支捕获。

import asyncio

async def worker(task_id: int, delay: float, fail: bool):
    await asyncio.sleep(delay)
    if fail:
        raise RuntimeError(f"任务 {task_id} 执行崩溃")
    return f"任务 {task_id} 成功"

async def main_task_group():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(worker(1, 0.5, fail=False))
            t2 = tg.create_task(worker(2, 0.2, fail=True))  # 此任务将最先失败
            t3 = tg.create_task(worker(3, 1.0, fail=False)) # 此任务将被自动取消
    except* RuntimeError as eg:
        # 使用 except* 捕获 ExceptionGroup 中的具体异常类型
        for exc in eg.exceptions:
            print(f"捕获到运行时异常: {exc}")

asyncio.run(main_task_group())

2. 结果汇总与降级：asyncio.gather(return_exceptions=True)

如果业务场景要求部分任务失败不影响其余任务继续完成（例如并发批量抓取、多个渠道查询），则可以使用 asyncio.gather() 并务必设置 return_exceptions=True。此时异常不会向上抛出中断程序，而是作为普通的返回值放置在结果列表中，供调用者通过 isinstance(item, Exception) 进行过滤与降级。

import asyncio

async def query_service(name: str, should_fail: bool):
    await asyncio.sleep(0.1)
    if should_fail:
        raise ConnectionResetError(f"无法连通 {name}")
    return f"{name} 数据"

async def main_gather():
    results = await asyncio.gather(
        query_service("服务A", False),
        query_service("服务B", True),
        query_service("服务C", False),
        return_exceptions=True
    )
    for idx, item in enumerate(results, 1):
        if isinstance(item, Exception):
            print(f"步骤 {idx} 发生异常，执行备用方案: {item}")
        else:
            print(f"步骤 {idx} 正常返回: {item}")

asyncio.run(main_gather())

四、异步取消（CancelledError）的处理禁忌与资源防泄露

在异步编程中，无论是 wait_for、timeout 还是 TaskGroup 取消未完成的任务，其底层核心机制都是向对应协程注入 asyncio.CancelledError。这是 Python 异步协程能够提前终止的根本原因，但也由此带来了极容易引发严重故障的处理禁忌。

import asyncio

async def correct_cleanup_worker():
    db_conn = await open_connection()
    try:
        while True:
            await perform_sync_work(db_conn)
    except asyncio.CancelledError:
        # 正确做法：记录日志或进行异步现场清理，但务必将其重新抛出
        print("收到外部取消指令，准备关闭连接...")
        raise
    finally:
        # 无论发生什么异常还是被取消，finally 块都会可靠执行
        await db_conn.close()
        print("连接安全释放")

另外，当协程执行到某些至关重要的非幂等操作（例如扣减账户余额、向消息队列提交确认偏移量）时，若中途被取消可能导致数据状态不一致。此时可以使用 asyncio.shield(coro) 保护关键逻辑。被 shield 保护的协程在外部发生超时或取消时，依然会在后台完整运行完毕，而不会被强行截断。

五、综合选型与实践准则

在项目架构中处理异步错误与超时，可根据下表快速匹配技术方案：