Python asyncio 如何优雅取消多个任务

Python asyncio 中的任务取消并非强制终止，而是通过抛出 CancelledError 异常，促使协程主动捕获、释放资源并完成清理——真正的“优雅取消”取决于协程是否响应这一信号；推荐使用 Python 3.11+ 的 `asyncio.TaskGroup`，它能自动统一管理任务生命周期，任一任务失败或取消都会触发全体清理并安全等待收尾；对于旧版本或需精细控制的场景，则需手动调用 `cancel()` 并配合 `gather(return_exceptions=True)` 协同处理，同时务必避开同步阻塞操作、避免静默吞掉取消异常，并在长耗时计算中主动轮询取消状态，才能确保异步程序健壮、可控、真正可取消。

理解任务取消的本质

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 密集型任务无法响应。

今天关于《Python asyncio 如何优雅取消多个任务》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！