Python中的协程任务取消与异常处理机制

1. 问题背景

在异步编程中，协程（Coroutine）可能因外部条件变化（如用户中断、超时等）需要被取消。取消任务时，需确保资源被正确释放，且异常能合理传递。Python的asyncio库提供了任务取消机制，但若处理不当，可能导致资源泄漏或程序崩溃。

2. 协程取消的基本原理

• 任务取消的触发：通过调用Task.cancel()方法，向协程发送CancelledError异常。
• 协程的响应：协程内部需捕获CancelledError并执行清理操作（如关闭文件、网络连接等），或选择忽略取消（通过shield保护）。
• 取消的状态流转：
  1. 任务进入cancelling状态。
  2. 若协程捕获CancelledError并重新抛出，任务变为cancelled；若协程忽略取消或处理异常后继续运行，任务恢复为running。

3. 取消操作的代码示例

步骤1：基本取消流程

import asyncio

async def simple_task():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("任务被取消！")
        raise  # 必须重新抛出CancelledError，否则任务不会进入cancelled状态

async def main():
    task = asyncio.create_task(simple_task())
    await asyncio.sleep(0.1)
    task.cancel()  # 发送取消请求
    try:
        await task
    except asyncio.CancelledError:
        print("任务已确认取消")

asyncio.run(main())

说明：

• 协程内必须重新抛出CancelledError，否则取消不会生效。
• 外层通过await task捕获CancelledError，确认任务已终止。

步骤2：资源清理与finally块

若协程持有资源（如打开的文件），需在finally块中释放：

async def task_with_cleanup():
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("取消中，正在清理资源...")
        raise
    finally:
        print("资源清理完成")  # 无论是否取消都会执行

步骤3：屏蔽取消（asyncio.shield）

某些关键操作不可被中断，可用asyncio.shield保护：

async def critical_task():
    await asyncio.sleep(2)
    print("关键操作完成")

async def main():
    task = asyncio.create_task(critical_task())
    shielded = asyncio.shield(task)  # 屏蔽取消
    shielded.cancel()  # 取消请求不会影响原task
    try:
        await shielded  # 这里会抛出CancelledError，但critical_task()继续运行
    except asyncio.CancelledError:
        print("shielded任务被取消，但原任务仍在运行")
    await task  # 等待原任务完成

注意：shield仅保护协程不被取消，但外层仍需处理CancelledError。

4. 取消与超时的结合

超时本质是自动取消，可通过asyncio.wait_for实现：

async def long_running_task():
    await asyncio.sleep(10)

async def main():
    try:
        await asyncio.wait_for(long_running_task(), timeout=1.0)
    except asyncio.TimeoutError:
        print("任务超时，自动取消")

底层机制：wait_for内部创建超时任务，超时后取消原任务。

5. 异常传递与调试

• 若协程内捕获CancelledError但未重新抛出，任务可能进入异常状态（如因其他异常失败）。
• 可通过Task.exception()获取任务异常：

async def faulty_task():
    try:
        await asyncio.sleep(1)
        raise ValueError("内部错误")
    except asyncio.CancelledError:
        print("取消被忽略，继续执行")
        # 未重新抛出CancelledError，任务继续运行并抛出ValueError

async def main():
    task = asyncio.create_task(faulty_task())
    task.cancel()
    await task
    if exception := task.exception():
        print(f"任务异常: {exception}")

6. 最佳实践总结

1. 始终在协程内处理CancelledError：清理资源后重新抛出异常。
2. 避免屏蔽取消后无限阻塞：被shield保护的任务需有超时机制。
3. 区分取消与其他异常：通过Task.cancelled()和Task.exception()判断任务终止原因。
4. 调试技巧：使用asyncio.debug=True启用详细日志，跟踪取消传播路径。

通过以上步骤，可确保协程取消行为符合预期，避免资源泄漏与状态不一致问题。