Python 3.11 asyncio.TaskGroup 实战:替代 gather 做并发,一个失败自动取消其余用asyncio.gather跑一批并发任务时,你可能踩过这些坑:某个任务抛异常了,其他任务还在后台傻跑;或者你用create_task手动起了一堆任务,忘了 await,结果收到一句刺眼的 “Task was destroyed but it is pending”。Python 3.11 引入的asyncio.TaskGroup就是来收拾这些烂摊子的——它是结构化并发的官方答案。这篇对比gather和TaskGroup,把「一个任务失败该怎么办」这件事讲清楚。先看 gather 的问题假设我们并发抓三个 URL,其中一个会失败:importasyncioasyncdeffetch(name,delay,failFalse):print(f{name}开始)awaitasyncio.sleep(delay)iffail:raiseValueError(f{name}挂了)print(f{name}完成)# 注意观察这行会不会打印returnnameasyncdefmain():try:awaitasyncio.gather(fetch(A,1),fetch(B,2,failTrue),# 1 秒后…其实是 2 秒后抛fetch(C,3),)exceptValueErrorase:print(捕获异常:,e)asyncio.run(main())输出:A 开始 B 开始 C 开始 A 完成 捕获异常: B 挂了看出问题了吗?B 在第 2 秒抛异常,gather立刻把异常抛给了main,但C 并没有被取消——它还在后台跑到第 3 秒。因为 C 的 “完成” 没打印出来只是程序在 B 异常后就结束了,如果main后面还有逻辑,C 会以「孤儿任务」的形式继续消耗资源。gather默认不会帮你取消兄弟任务。你可能会说加return_exceptionsTrue?那只是把异常当普通返回值收集起来,任务还是各跑各的,更不会互相取消。换成 TaskGroupTaskGroup用async with管理一组任务的生命周期:退出async with块时,它会等所有任务结束;只要有一个任务抛异常,其余未完成的任务会被自动取消。importasyncioasyncdeffetch(name,delay,failFalse):try:print(f{name}开始)awaitasyncio.sleep(delay)iffail:raiseValueError(f{name}挂了)print(f{name}完成)returnnameexceptasyncio.CancelledError:# 被兄弟任务的异常连累取消时,会走到这里print(f{name}被取消)raiseasyncdefmain():try:asyncwithasyncio.TaskGroup()astg:tg.create_task(fetch(A,1))tg.create_task(fetch(B,2,failTrue))tg.create_task(fetch(C,3))except*ValueErroraseg:# 注意这里是 except*,捕获的是 ExceptionGroupprint(捕获异常组:,eg.exceptions)asyncio.run(main())输出:A 开始 B 开始 C 开始 A 完成 C 被取消 捕获异常组: (ValueError(B 挂了),)区别一目了然:B 在第 2 秒挂掉后,C 立刻收到CancelledError被取消,不再空跑到第 3 秒。A 因为第 1 秒就已经完成,不受影响。这才是「一荣俱荣做完、一损则止损」的结构化并发。关键点一:异常是 ExceptionGroup,要用 except*注意上面用的是except* ValueError(带星号),而不是普通的except。因为 TaskGroup 里可能同时有多个任务失败,它会把这些异常打包成一个ExceptionGroup一起抛出。except*(Python 3.11 同步引入)专门用来解构 ExceptionGroup:asyncdefmain():try:asyncwithasyncio.TaskGroup()astg:tg.create_task(fail_with(ValueError(v)))tg.create_task(fail_with(KeyError(k)))except*ValueErroraseg:print(值错误们:,eg.exceptions)except*KeyErroraseg:print(键错误们:,eg.exceptions)两个任务同时挂,两个except*分支都会命中。如果你用老的except ValueError,面对 ExceptionGroup 是捕获不到的——这是从 gather 迁移过来最容易懵的地方。关键点二:create_task 必须在 async with 块内tg.create_task()只能在async with tg:的作用域内部调用。一旦退出这个块,TaskGroup 就「关闭」了,再往里加任务会抛RuntimeError。这其实是好事——它从语法上保证了「所有任务都在同一个作用域里被管理和等待」,不会有漏网的孤儿任务。asyncdefmain():asyncwithasyncio.TaskGroup()astg:tasks[tg.create_task(fetch(ft{i},i*0.1))foriinrange(5)]# 出了 async with,所有任务保证已完成# 此时才能安全地读结果print([t.result()fortintasks])想拿返回值,就先把create_task返回的 Task 对象存下来,等async with退出后再.result()。这时所有任务一定都结束了,.result()不会阻塞。关键点三:限流仍然要靠 SemaphoreTaskGroup 负责生命周期,但不负责限流——你create_task100 个,它就同时跑 100 个。要控制并发度,还是老搭档Semaphore:asyncdeffetch_limited(sem,name,delay):asyncwithsem:# 最多 3 个同时进入awaitasyncio.sleep(delay)returnnameasyncdefmain():semasyncio.Semaphore(3)# 并发上限 3asyncwithasyncio.TaskGroup()astg:tasks[tg.create_task(fetch_limited(sem,ft{i},1))foriinrange(10)]print([t.result()fortintasks])10 个任务、上限 3,分批跑完,同时享受 TaskGroup 的自动取消和异常聚合。二者职责分明:Semaphore 管「同时几个」,TaskGroup 管「怎么收尾」。小结gather的短板:某任务失败后不会取消兄弟任务,容易留下后台空跑的孤儿任务;return_exceptionsTrue也只是收集异常,不改变这点。TaskGroup(3.11)用async with做结构化并发:任一任务异常 → 自动取消其余,退出块时保证全部结束。异常被打包成ExceptionGroup,必须用except*解构,普通except抓不到——这是迁移第一坑。create_task只能在async with块内调用;想拿结果就存 Task,出块后再.result()。TaskGroup 不管限流,并发度控制仍交给Semaphore。一句话记忆:gather 是「各跑各的」,TaskGroup 是「一损俱止损」——要结构化并发和自动取消,就用 TaskGroup except。* Python 3.11 asyncio.TaskGroup 实战:替代 gather 做并发,一个失败自动取消其余
Python 3.11 asyncio.TaskGroup 实战:替代 gather 做并发,一个失败自动取消其余用asyncio.gather跑一批并发任务时,你可能踩过这些坑:某个任务抛异常了,其他任务还在后台傻跑;或者你用create_task手动起了一堆任务,忘了 await,结果收到一句刺眼的 “Task was destroyed but it is pending”。Python 3.11 引入的asyncio.TaskGroup就是来收拾这些烂摊子的——它是结构化并发的官方答案。这篇对比gather和TaskGroup,把「一个任务失败该怎么办」这件事讲清楚。先看 gather 的问题假设我们并发抓三个 URL,其中一个会失败:importasyncioasyncdeffetch(name,delay,failFalse):print(f{name}开始)awaitasyncio.sleep(delay)iffail:raiseValueError(f{name}挂了)print(f{name}完成)# 注意观察这行会不会打印returnnameasyncdefmain():try:awaitasyncio.gather(fetch(A,1),fetch(B,2,failTrue),# 1 秒后…其实是 2 秒后抛fetch(C,3),)exceptValueErrorase:print(捕获异常:,e)asyncio.run(main())输出:A 开始 B 开始 C 开始 A 完成 捕获异常: B 挂了看出问题了吗?B 在第 2 秒抛异常,gather立刻把异常抛给了main,但C 并没有被取消——它还在后台跑到第 3 秒。因为 C 的 “完成” 没打印出来只是程序在 B 异常后就结束了,如果main后面还有逻辑,C 会以「孤儿任务」的形式继续消耗资源。gather默认不会帮你取消兄弟任务。你可能会说加return_exceptionsTrue?那只是把异常当普通返回值收集起来,任务还是各跑各的,更不会互相取消。换成 TaskGroupTaskGroup用async with管理一组任务的生命周期:退出async with块时,它会等所有任务结束;只要有一个任务抛异常,其余未完成的任务会被自动取消。importasyncioasyncdeffetch(name,delay,failFalse):try:print(f{name}开始)awaitasyncio.sleep(delay)iffail:raiseValueError(f{name}挂了)print(f{name}完成)returnnameexceptasyncio.CancelledError:# 被兄弟任务的异常连累取消时,会走到这里print(f{name}被取消)raiseasyncdefmain():try:asyncwithasyncio.TaskGroup()astg:tg.create_task(fetch(A,1))tg.create_task(fetch(B,2,failTrue))tg.create_task(fetch(C,3))except*ValueErroraseg:# 注意这里是 except*,捕获的是 ExceptionGroupprint(捕获异常组:,eg.exceptions)asyncio.run(main())输出:A 开始 B 开始 C 开始 A 完成 C 被取消 捕获异常组: (ValueError(B 挂了),)区别一目了然:B 在第 2 秒挂掉后,C 立刻收到CancelledError被取消,不再空跑到第 3 秒。A 因为第 1 秒就已经完成,不受影响。这才是「一荣俱荣做完、一损则止损」的结构化并发。关键点一:异常是 ExceptionGroup,要用 except*注意上面用的是except* ValueError(带星号),而不是普通的except。因为 TaskGroup 里可能同时有多个任务失败,它会把这些异常打包成一个ExceptionGroup一起抛出。except*(Python 3.11 同步引入)专门用来解构 ExceptionGroup:asyncdefmain():try:asyncwithasyncio.TaskGroup()astg:tg.create_task(fail_with(ValueError(v)))tg.create_task(fail_with(KeyError(k)))except*ValueErroraseg:print(值错误们:,eg.exceptions)except*KeyErroraseg:print(键错误们:,eg.exceptions)两个任务同时挂,两个except*分支都会命中。如果你用老的except ValueError,面对 ExceptionGroup 是捕获不到的——这是从 gather 迁移过来最容易懵的地方。关键点二:create_task 必须在 async with 块内tg.create_task()只能在async with tg:的作用域内部调用。一旦退出这个块,TaskGroup 就「关闭」了,再往里加任务会抛RuntimeError。这其实是好事——它从语法上保证了「所有任务都在同一个作用域里被管理和等待」,不会有漏网的孤儿任务。asyncdefmain():asyncwithasyncio.TaskGroup()astg:tasks[tg.create_task(fetch(ft{i},i*0.1))foriinrange(5)]# 出了 async with,所有任务保证已完成# 此时才能安全地读结果print([t.result()fortintasks])想拿返回值,就先把create_task返回的 Task 对象存下来,等async with退出后再.result()。这时所有任务一定都结束了,.result()不会阻塞。关键点三:限流仍然要靠 SemaphoreTaskGroup 负责生命周期,但不负责限流——你create_task100 个,它就同时跑 100 个。要控制并发度,还是老搭档Semaphore:asyncdeffetch_limited(sem,name,delay):asyncwithsem:# 最多 3 个同时进入awaitasyncio.sleep(delay)returnnameasyncdefmain():semasyncio.Semaphore(3)# 并发上限 3asyncwithasyncio.TaskGroup()astg:tasks[tg.create_task(fetch_limited(sem,ft{i},1))foriinrange(10)]print([t.result()fortintasks])10 个任务、上限 3,分批跑完,同时享受 TaskGroup 的自动取消和异常聚合。二者职责分明:Semaphore 管「同时几个」,TaskGroup 管「怎么收尾」。小结gather的短板:某任务失败后不会取消兄弟任务,容易留下后台空跑的孤儿任务;return_exceptionsTrue也只是收集异常,不改变这点。TaskGroup(3.11)用async with做结构化并发:任一任务异常 → 自动取消其余,退出块时保证全部结束。异常被打包成ExceptionGroup,必须用except*解构,普通except抓不到——这是迁移第一坑。create_task只能在async with块内调用;想拿结果就存 Task,出块后再.result()。TaskGroup 不管限流,并发度控制仍交给Semaphore。一句话记忆:gather 是「各跑各的」,TaskGroup 是「一损俱止损」——要结构化并发和自动取消,就用 TaskGroup except。*