Python中的异步任务取消传播与异常处理链

题目描述：
在异步编程中，任务取消（Cancellation）和异常处理是核心机制。当异步任务被取消时，取消信号如何在协程链中传播？异常在异步调用链中如何传递？asyncio.CancelledError 与普通异常有何区别？asyncio.shield() 如何保护任务不被取消？这些机制的理解对编写健壮的异步程序至关重要。

解题过程：

1. 异步任务取消的基本机制

在 asyncio 中，取消是通过调用 Task.cancel() 触发的。取消操作并不会强制终止任务，而是向任务发送一个取消请求：

• 任务在执行到下一个 await 点时，会抛出 asyncio.CancelledError 异常。
• 该异常必须被协程捕获并处理，否则任务会静默结束。

示例：

import asyncio

async def my_task():
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("任务被取消！")
        raise  # 通常重新抛出，确保任务状态变为取消

async def main():
    task = asyncio.create_task(my_task())
    await asyncio.sleep(1)  # 等待1秒
    task.cancel()           # 发送取消请求
    await task              # 等待任务结束

asyncio.run(main())

输出：

任务被取消！

这里，my_task 在 await asyncio.sleep(5) 时收到取消信号，抛出 CancelledError，被捕获后打印消息，再重新抛出以确认取消。

2. 取消信号的传播路径

当异步任务包含子协程调用时，取消信号会沿调用链传播：

• 如果父协程被取消，CancelledError 会从当前 await 点抛出。
• 如果子协程正在执行，取消信号会传递给子协程，子协程也会收到 CancelledError。

示例：

async def child():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("子协程被取消")
        raise

async def parent():
    try:
        await child()
    except asyncio.CancelledError:
        print("父协程被取消")
        raise

async def main():
    task = asyncio.create_task(parent())
    await asyncio.sleep(1)
    task.cancel()
    await task

asyncio.run(main())

输出：

子协程被取消
父协程被取消

取消信号从 parent 传递到 child，两者依次被取消。

3. 异常在异步链中的传递

异步调用链中，异常传递与同步代码类似：

• 子协程抛出的异常会传递给父协程的 await 点。
• 如果异常未被捕获，会导致任务失败。

示例：

async def child():
    raise ValueError("子协程错误")

async def parent():
    try:
        await child()
    except ValueError as e:
        print(f"父协程捕获: {e}")

asyncio.run(parent())

输出：

父协程捕获: 子协程错误

4. CancelledError 的特殊性

asyncio.CancelledError 是 BaseException 的子类（而非 Exception 的子类），因此：

• 不会被普通的 except Exception: 捕获。
• 这种设计确保取消信号不会被意外吞没。

示例：

async def task():
    try:
        await asyncio.sleep(5)
    except Exception:  # 不会捕获 CancelledError！
        print("捕获异常")

async def main():
    t = asyncio.create_task(task())
    await asyncio.sleep(1)
    t.cancel()
    await t
    print(t.cancelled())  # True

asyncio.run(main())

输出：

True

由于 CancelledError 未被捕获，任务直接取消，不会打印 "捕获异常"。

5. 保护任务不被取消：asyncio.shield()

shield() 可以保护一个协程不被取消，但行为较微妙：

• 被 shield() 包裹的协程仍可被取消，但取消信号不会立即生效。
• 外部调用者仍会收到 CancelledError，但被保护的协程会继续运行，直到完成。

示例：

async def protected():
    await asyncio.sleep(3)
    return "完成"

async def main():
    task = asyncio.create_task(protected())
    shielded = asyncio.shield(task)
    await asyncio.sleep(1)
    shielded.cancel()  # 取消shielded包装
    try:
        await shielded
    except asyncio.CancelledError:
        print("shielded被取消")
    # 原始任务继续执行
    result = await task
    print(f"原始任务结果: {result}")

asyncio.run(main())

输出：

shielded被取消
原始任务结果: 完成

shield() 仅保护原始任务 task 不被取消，但 shielded 这个包装对象会立即响应取消。

6. 正确处理取消的实践

1. 资源清理：在 except asyncio.CancelledError 中释放资源（如关闭网络连接）。
2. 重新抛出：除非明确要忽略取消，否则应重新抛出 CancelledError。
3. 超时与取消：asyncio.wait_for() 在超时时会取消内部任务，需注意清理。

示例：

async def cleanup_on_cancel():
    resource = "打开的资源"
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print(f"清理 {resource}")
        raise
    finally:
        print("finally块始终执行")

总结：

• 取消通过 CancelledError 在异步链中传播，需显式处理。
• 异常传递与同步代码类似，但 CancelledError 是 BaseException 子类。
• shield() 可延迟取消，但不完全免疫取消。
• 健壮代码应在取消时进行资源清理，并通常重新抛出 CancelledError。