Python TaskGroup 失败与取消：让并发任务结束得可验证
一项请求同时启动几项相关工作时，最难处理的往往不是全部成功，而是其中一项失败后，其余任务还在运行。Python 3.11 引入的 TaskGroup 把这些任务收进同一个异步上下文：退出时等待组内任务结束，并把失败带回调用处。本文示例适用于 Python 3.11 及以上，只在内存中调度协程，不访问网络、不改动真实服务。
AI生成概念配图：同一边界内的并行任务在一处失败后收束，并完成清理。仅作概念说明，不代表实际界面或实测结果。
先区分失败和取消
组内任务抛出普通异常时，TaskGroup 会请求取消其余尚未完成的任务，等待它们收尾，再抛出包含失败的异常组。单个任务被取消则不会自动把其他成员一起取消。因此，业务失败与主动停止某项工作应保留不同语义，不能都转换成一个空结果。已完成的任务也不会因为同伴失败而被撤销。
下面用事件确认等待任务已经启动，再让另一项任务失败，避免依赖“睡几毫秒应该够了”的时序猜测。把完整代码保存为 taskgroup_demo.py 后用 python3 运行。第二段场景单独取消等待任务，并验证同组另一项任务仍正常返回。
import asyncio
async def waiting(ready, events):
try:
ready.set()
await asyncio.Event().wait()
except asyncio.CancelledError:
events.append('worker:cancelled')
raise
finally:
events.append('worker:cleanup')
async def failing(ready):
await ready.wait()
raise ValueError('sample failure')
async def failed_group():
ready = asyncio.Event()
events = []
try:
async with asyncio.TaskGroup() as group:
worker = group.create_task(waiting(ready, events))
group.create_task(failing(ready))
except* ValueError as errors:
events.append(f'caught:{len(errors.exceptions)}')
assert worker.cancelled()
assert events == ['worker:cancelled', 'worker:cleanup', 'caught:1']
print(*events, sep='\n')
async def one_cancelled():
ready = asyncio.Event()
events = []
async with asyncio.TaskGroup() as group:
worker = group.create_task(waiting(ready, events))
survivor = group.create_task(
asyncio.sleep(0, result='sibling:finished'))
await ready.wait()
worker.cancel()
assert worker.cancelled()
assert survivor.result() == 'sibling:finished'
assert events == ['worker:cancelled', 'worker:cleanup']
print(survivor.result())
async def main():
await failed_group()
await one_cancelled()
asyncio.run(main())退出上下文前，清理已经执行
第一段依次打印 worker:cancelled、worker:cleanup、caught:1；第二段打印 sibling:finished。断言同时检查取消状态、清理记录以及正常结果。本次在 Python 3.12.14 实际运行通过。这些输出只描述此处受控场景，不保证任意并发程序中的日志总以相同顺序出现。
finally 是释放本任务持有资源的位置。若显式捕获 CancelledError，完成必要清理后通常应重新抛出，让上层知道任务确实被取消。它继承自 BaseException，普通 except Exception 不会捕获它；为了“避免报错”而笼统捕获 BaseException，反而容易破坏取消和退出流程。
异常组要按类型处理
except* ValueError 处理异常组中匹配的部分，其他类型仍向外传播。示例只有一个直接失败，所以打印数量为一；嵌套任务组会形成嵌套异常结构，不能把顶层 exceptions 的长度当作所有根因的总数。实际排查应保留原始异常组及堆栈，并在能够恢复的层次处理特定失败。
不要在捕获后立即读取每个任务的 result()：被取消任务会再次抛出取消异常，失败任务会抛出原来的失败。只有成功退出任务组、确认相关任务未取消且无异常后，才能把结果当作完整成功批次使用。若产品允许部分成功，需要显式定义结果对象、失败类别和重试规则。
任务必须通过这个组的 create_task 加入管理。在组内随手调用全局 asyncio.create_task，不会自动成为成员，任务组也不会替你等待那个后台任务。审查代码时应沿着任务创建位置检查归属，避免主流程退出后仍留下没有负责人等待的工作。
取消不是强制终止，也不是回滚
cancel() 发出请求，协程要到能够响应取消的位置才会收到异常。长时间同步计算或阻塞调用会拖住事件循环；清理过程过长也会延迟任务组退出。示例清理只追加内存记录，真实资源清理应有明确边界，不能无限等待或吞掉取消。
任务组不提供数据库事务，不能撤销已经发送的消息、写入的文件或其他外部副作用。需要重试时，应另行设计幂等性和补偿。上线前至少覆盖全部成功、单项失败、主动取消、外部取消以及清理失败；嵌套组和同时取消的细节还应在项目实际使用的 Python 小版本上验证。
此外，任务组负责生命周期，不会自动限制同时发起的请求数量。把整份大清单一次创建成任务，仍可能耗尽连接或内存；并发上限应由调用方按资源预算另行控制。
参考资料
资料核验日期：2026年9月30日，UTC。
