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
<p>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 知道已处理</p><p>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("主流程也被取消")</p><p>asyncio.run(main())立即学习“Python免费学习笔记（深入）”；
手动管理时用 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 密集型任务无法响应。
