Python asyncio 如何优雅取消正在运行的多个相关任务

asyncio任务取消本质是抛出CancelledError让协程主动退出；需捕获异常、释放资源、完成收尾；TaskGroup（Python 3.11+）自动统一取消并等待清理。

理解任务取消的本质

asyncio 中取消任务不是强制终止，而是通过抛出 CancelledError 让协程主动退出。真正“优雅”的关键在于：协程需捕获该异常、释放资源（如关闭连接、清理临时文件）、并完成必要的收尾逻辑。若任务内部阻塞在未响应取消的 IO（如某些同步库调用）或忽略异常，则无法真正取消。

使用 asyncio.TaskGroup 统一管理（推荐，Python 3.11+）

TaskGroup 是目前最简洁、安全的方式——它自动绑定子任务生命周期，任一任务失败或被取消，其余任务会收到统一取消信号，并等待它们完成清理。

• 用 async with asyncio.TaskGroup() as tg: 创建上下文
• 所有通过 tg.create_task() 启动的任务共享同一个取消语义
• 外部调用 tg.cancel() 或上下文退出时异常传播，都会触发全部任务的取消流程

示例：

import asyncio
async def worker(name: str):
try:
await asyncio.sleep(3)
print(f"{name} 完成")
except asyncio.CancelledError:
print(f"{name} 被取消，正在清理...")
await asyncio.sleep(0.5)  # 模拟清理耗时
print(f"{name} 清理完毕")
raise  # 重新抛出，确保 TaskGroup 知道已处理
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(worker("A"))
tg.create_task(worker("B"))
tg.create_task(worker("C"))
await asyncio.sleep(1)  # 运行 
1 秒后取消
tg.cancel()  # 主动取消全部
except asyncio.CancelledError:
print("主流程也被取消")
asyncio.run(main())

手动管理时用 cancel() + gather(return_exceptions=True)

对于 Python

• 保存所有 asyncio.Task 实例到列表
• 对每个 task 调用 .cancel()
• 用 await asyncio.gather(*tasks, return_exceptions=True) 等待全部完成（避免因某个任务抛 CancelledError 导致其他任务被中断）
• 检查每个结果：是 CancelledError 表示已取消；是其他异常需单独处理；是正常返回值说明成功完成

避免常见陷阱

不要在协程中静默吞掉 CancelledError —— 若只写 except Exception: 或空 except:，取消信号会被吃掉，任务卡住。
慎用 time.sleep() / threading.Lock / requests.get() —— 这些是同步阻塞操作，不响应 asyncio 取消。应改用 asyncio.sleep()、异步锁（asyncio.Lock）、aiohttp 等原生异步库。
长时间计算循环需主动检测取消状态 —— 在循环体内定期调用 if asyncio.current_task().cancelled(): raise asyncio.CancelledError，防止 CPU 密集型任务无法响应。