taskgroup 是 python 3.11 推荐替代 asyncio.gather() 的并发控制方式，核心优势在于错误原样抛出、任一失败即终止全部任务，并支持精确异常捕获与上下文感知取消；适用于需降级处理、任务依赖或库函数错误区分的场景。
TaskGroup 是 Python 3.11 引入的正式替代 asyncio.gather() 和手动 asyncio.create_task() + await 的首选方式，但它不是“万能胶”——用错场景反而会让错误更难调试。
什么时候必须用 TaskGroup 而不是 asyncio.gather()
核心区别在错误传播和生命周期控制：gather() 默认“全等”（所有任务完成才返回），出错时默认取消其余任务但不抛出；TaskGroup 则强制“任一失败即终止全部”，且错误原样抛出，支持 except 捕获。
- 需要精确捕获某子任务的异常（比如 HTTP 请求失败后做降级处理）→ 必须用
TaskGroup - 子任务之间有依赖或需共享取消信号（如一个超时导致全部退出）→
TaskGroup自带上下文感知取消 - 你正在写库函数，希望调用方能用
try/except区分不同子任务的错误类型 →TaskGroup保留原始异常栈 - 只是并发跑几个独立 IO 操作、不关心谁先失败 →
gather(return_exceptions=True)更轻量
TaskGroup 中如何安全地传递状态或共享资源
不能直接在 async with asyncio.TaskGroup() as tg: 块外访问任务对象，也不能靠闭包变量同步状态——因为子任务是并发执行的，没有隐式锁。
- 用
asyncio.Lock保护共享变量（如计数器、缓存字典） - 传参只通过函数参数，避免修改外层变量：错误写法
results.append(...)→ 正确写法是让每个子协程返回值，由TaskGroup收集到task.result() - 若需跨任务通信，优先用
asyncio.Queue，而非全局列表或threading.local - 注意：
TaskGroup不提供类似concurrent.futures.as_completed()的“按完成顺序获取结果”能力，需自行用asyncio.Queue或asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)
常见报错：RuntimeError: Task group was cancelled 怎么定位
这不是你的代码错了，而是 TaskGroup 在退出时检测到有子任务未完成或已被取消——通常因为子任务内部没处理好取消信号。
- 检查子协程是否用了阻塞调用（如
time.sleep()、requests.get()），它们不会响应asyncio.CancelledError - 子协程里做了长时间 CPU 计算？加
await asyncio.sleep(0)让出控制权，否则无法被及时取消 - 子协程 await 了另一个未设 timeout 的协程（如没加
asyncio.wait_for(..., timeout=...)的网络请求） - 最简单的复现方式：
async def bad(): await asyncio.sleep(10); return "done"→ 在TaskGroup中启动它然后提前退出上下文
性能与兼容性要注意的点
TaskGroup 本身开销极小，但它的语义约束会暴露原有异步逻辑的缺陷。
- Python anyio 的
create_task_group），行为不一致 - 不能嵌套使用同一个
TaskGroup实例，但可以多层嵌套不同实例（外层 cancel 会透传到内层） - 和
asyncio.to_thread()混用时，线程中抛出的异常会被包装成asyncio.CancelledError，需在外层用except BaseException捕获 - 如果你原来用
gather()并发 100 个任务，改用TaskGroup后发现内存涨了一截：大概率是子协程里创建了大量未释放的引用（比如闭包捕获了大对象），TaskGroup的严格生命周期让这类问题更早暴露
真正难的不是写对 TaskGroup 的语法，而是想清楚：哪些子任务该归为一组？它们失败时是否应该互相拖垮？有没有隐藏的资源竞争没被锁住？这些问题不厘清，换什么语法都救不了逻辑。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
