手上有个抓取任务，每次要拉几百个商品的价格，用的是很常规的写法：把请求拼成协程列表，交给 asyncio.gather 并发跑，跑完写库。日志里一直显示“抓取完成 320 条”，直到有次对账发现少了十几条，翻回去看日志，确实没有报错，也没有任何异常堆栈。
问题出在 gather 的 return_exceptions=True 上。它把异常当成正常返回值塞进了结果列表，代码拿到列表就往下走，没有逐条判断，于是失败的请求被当成空数据一路带到了入库环节。这不是什么高深的问题，但 gather 的默认行为确实容易让人写出“看着没事”的代码。
后来我把这段逻辑用 TaskGroup 重写了一遍，顺手补上了限流、重试和超时。这篇把全过程记下来，代码在 JDK、Python 3.11 以上都能直接跑。
gather 让人放松警惕的三个地方
第一个是异常传播。return_exceptions=False 时，只要有一个协程抛异常，gather 会立刻把异常往上抛，但剩下的协程不会被取消，它们还在后台跑，跑完的结果没人收。如果在接口里这么写，用户看到的是 500，但后台还在跑几十秒的抓取任务，对下游是持续的无效压力。
第二个是异常吞没。为了避免上面那种情况，大家会习惯性加 return_exceptions=True，结果就是异常变成了返回值。返回列表里混着 dict 和 Exception，判不判全凭自觉。一旦漏判，失败的请求就成了静默丢失。
第三个是取消不可控。gather 返回的 Future 上调用 cancel()，按文档，只是取消那个外层的聚合 Future，里面的子任务不一定会跟着停。在请求超时的场景里，这意味着你以为已经放弃了，实际后台还挂着一堆网络调用。
这三个问题本质上是同一个：gather 只负责“把协程并行跑起来”，不负责“这些协程之间是什么关系、失败之后怎么办”。它没有一个明确的作用域概念，失败和取消都没有统一的处理点。
TaskGroup 补上的就是这块
Python 3.11 加的 asyncio.TaskGroup 做了两件关键的事。
第一，退出 with 块时，保证所有子任务都已经结束，要么完成，要么被取消。不会留下“还在后台跑”的孤儿任务。
第二，任何一个子任务抛异常，其余任务会立即被取消，异常被收集到一个 ExceptionGroup 里，在退出 with 块时统一抛出。这就把“失败即整体失败”做成了默认行为，不用手动去传取消信号。
当然，抓取场景里未必希望“一个失败全部取消”。如果 320 个商品里有一个接口报 500，我们希望的是那一条失败、其余继续。这种需求也不能靠放松异常处理来做，正确方式是在每个任务内部把“单条失败”消化掉、记下错误，只让真正不可恢复的错误往外冒。后面的代码会体现这一点。
先把问题复现出来
写一个模拟的下游接口，一部分请求会失败，延迟也不一样，方便观察取消和异常的行为。
import asyncio
import random
import time
async def fetch_price(item_id: int) -> dict:
"""模拟下游价格接口：30% 概率超时失败，其余返回价格"""
await asyncio.sleep(random.uniform(0.05, 0.35))
if item_id % 7 == 0:
raise ConnectionError(f"item {item_id} 请求超时")
return {"id": item_id, "price": item_id * 1.5}
然后用最朴素的 gather 写法跑一遍，注意不要加 return_exceptions=True，看看会发生什么。
async def naive_gather():
tasks = [fetch_price(i) for i in range(1, 51)]
start = time.perf_counter()
try:
results = await asyncio.gather(*tasks)
print(f"全部成功，共 {len(results)} 条，耗时 {time.perf_counter() - start:.2f}s")
except ConnectionError as e:
print(f"第一个异常：{e}")
print(f"gather 退出耗时 {time.perf_counter() - start:.2f}s")
print(f"当前剩余任务数：{len(asyncio.all_tasks()) - 1}")
asyncio.run(naive_gather())
输出大概是这样的：
第一个异常：item 7 请求超时
gather 退出耗时 0.32s
当前剩余任务数：47
第一个异常在 0.3 秒左右冒出来了，但还有四十多个协程在后台跑。这就是前面说的第一个坑。如果这个函数是被一个请求处理器调用的，用户的请求已经返回了，后台还在占着连接池和事件循环的调度时间。
加上 return_exceptions=True 再看，异常确实被收进列表了，但结果里混着 Exception 对象。原来那段代码就是用 for r in results: db.insert(r) 塞进库的，此时塞进去的是一条 ConnectionError。数据库那层一般不校验类型，就直接进去了，等到对账的时候才发现。
用 TaskGroup 重写
重写的核心是先定义好“单条任务怎么算成功、怎么算失败”。我的约定是：
- 拿不到数据时，记录错误并返回 None，不往外抛。
- 只有内部出现编程错误（KeyError、TypeError 这类）才让它冒出去，让 TaskGroup 整体失败。
这样 TaskGroup 的“失败即取消”就只在真正该炸的时候触发，正常运行时的偶发失败不牵连其他任务。
async def fetch_one(item_id: int, sem: asyncio.Semaphore, errors: list):
async with sem:
try:
return await fetch_price(item_id)
except ConnectionError as e:
errors.append((item_id, str(e)))
return None
except Exception as e:
# 非预期错误重新抛出，让 TaskGroup 介入
raise
async def with_task_group(item_ids):
sem = asyncio.Semaphore(10)
errors: list = []
results: list = []
async with asyncio.TaskGroup() as tg:
tasks = [
tg.create_task(fetch_one(i, sem, errors), name=f"price-{i}")
for i in item_ids
]
# 到这里所有任务都已结束，不会遗留后台协程
for t in tasks:
r = t.result()
if r is not None:
results.append(r)
return results, errors
注意最后两件事。with 块结束之后，所有任务都一定已经 completed 或者 cancelled，所以循环拿 t.result() 是安全的。Semaphore 则把并发压到 10，避免一次性把下游打满。
如果某个任务抛了未捕获的异常，会走 ExceptionGroup，写法是这样的：
try:
results, errors = await with_task_group(range(1, 51))
except* KeyError as eg:
print(f"内部数据结构异常，涉及 {len(eg.exceptions)} 个任务")
except* ValueError as eg:
print(f"参数异常：{eg.exceptions}")
except* 是 3.11 一起加进来的语法，专门用来拆 ExceptionGroup。如果你还在用 3.10 或更早的版本，这一步做不到，这也是我没法把方案落到旧版本上的真实原因——不是不想，语法层面就不支持。
加上重试和退避
单条 fetch 失败一次就放弃，对偶发的网络抖动太苛刻。加一层带上限的重试，指数退避，随机抖动，这些都是异步任务里的常规做法。
async def fetch_with_retry(item_id: int, sem: asyncio.Semaphore, errors: list, max_attempts: int = 3):
async with sem:
last_err = None
for attempt in range(1, max_attempts + 1):
try:
return await fetch_price(item_id)
except ConnectionError as e:
last_err = e
if attempt == max_attempts:
break
backoff = 0.2 * (2 ** (attempt - 1)) * (0.5 + random.random())
await asyncio.sleep(backoff)
errors.append((item_id, str(last_err)))
return None
几个细节需要说。退避的随机因子很重要，没有抖动，一批任务会在同一时刻重新发起请求，正好把下游又打一波。重试次数不要设太高，三次已经是上限，一次抓取任务里每条都重试三次，总耗时会被最慢的那条拖长。
另外，重试的 await asyncio.sleep() 是在持有 semaphore 的情况下做的。这一点其实有争议：如果退避时还占着名额，其他任务就无法进入，整体吞吐会下降；但如果不占着名额，退避期间的重试请求可能和其他正常请求撞车。我的选择是在超时较短的场景（单个请求 5 秒以内、退避最多 1 秒）里保持占着名额，逻辑更简单、限流也更严格。如果退避时间超过 2 秒，就该把 sleep 挪到 semaphore 外面。
加上超时
3.11 还加了 asyncio.timeout，比 asyncio.wait_for 更好用，可以包住整个 TaskGroup。
async def run_batch(item_ids, timeout_sec: float = 20.0):
sem = asyncio.Semaphore(10)
errors: list = []
try:
async with asyncio.timeout(timeout_sec):
async with asyncio.TaskGroup() as tg:
tasks = [
tg.create_task(fetch_with_retry(i, sem, errors), name=f"price-{i}")
for i in item_ids
]
except TimeoutError:
print(f"整批抓取超过 {timeout_sec}s，已被取消")
return [], errors
results = [t.result() for t in tasks if t.result() is not None]
return results, errors
这里 TimeoutError 是从 asyncio.timeout 里抛出来的内置 TimeoutError，不是 asyncio.TimeoutError（旧名，3.11 之后两者已合并）。超时之后，TaskGroup 会把所有子任务取消，sem 的持有者在取消异常中退出 async with 块，semaphore 正常释放。如果某个任务正在 await asyncio.sleep() 里，这个取消也是安全的。
时间限制该给多少，取决于 item_ids 的长度和并发度。粗略算一下：10 并发、单个请求 p99 大概 0.5 秒、每条最多重试两次，320 个任务的时间下限大概是 320 / 10 × 0.5 × 3 ≈ 48 秒。所以直接给 20 秒是不够的。这种长度的抓取，正确做法是先估算下限，再留 30% 的余量。如果算出来发现时间太长，那就得提高并发度，或者把批量拆小，而不是压缩超时。
完整可运行版本
import asyncio
import random
import time
async def fetch_price(item_id: int) -> dict:
await asyncio.sleep(random.uniform(0.05, 0.35))
if item_id % 7 == 0:
raise ConnectionError(f"item {item_id} 请求超时")
return {"id": item_id, "price": round(item_id * 1.5, 2)}
async def fetch_with_retry(item_id, sem, errors, max_attempts=3):
async with sem:
last_err = None
for attempt in range(1, max_attempts + 1):
try:
return await fetch_price(item_id)
except ConnectionError as e:
last_err = e
if attempt == max_attempts:
break
backoff = 0.2 * (2 ** (attempt - 1)) * (0.5 + random.random())
await asyncio.sleep(backoff)
errors.append((item_id, str(last_err)))
return None
async def run_batch(item_ids, concurrency=10, timeout_sec=20.0):
sem = asyncio.Semaphore(concurrency)
errors: list = []
start = time.perf_counter()
try:
async with asyncio.timeout(timeout_sec):
async with asyncio.TaskGroup() as tg:
tasks = [
tg.create_task(fetch_with_retry(i, sem, errors), name=f"price-{i}")
for i in item_ids
]
except TimeoutError:
print(f"整批超时，已取消剩余任务，耗时 {time.perf_counter() - start:.2f}s")
return [], errors
results = [t.result() for t in tasks if t.result() is not None]
elapsed = time.perf_counter() - start
print(f"完成 {len(results)} 条，失败 {len(errors)} 条，耗时 {elapsed:.2f}s")
if errors:
print(f"失败示例：{errors[:3]}")
return results, errors
if __name__ == "__main__":
ids = list(range(1, 321))
results, errors = asyncio.run(run_batch(ids, concurrency=10, timeout_sec=40.0))
print(f"最终入库 {len(results)} 条，未成功 {len(errors)} 条")
跑下来大约 8 到 12 秒，320 条里能成功 270 多条，失败的都落在 errors 里，能定位到具体是哪几个 item 报超时。这个结果比原来的“看着成功，其实悄悄丢了十几条”好很多。
替换成真实请求
把 fetch_price 换成真实的 HTTP 调用，只需要改这一个函数。用 httpx，因为它支持连接池复用，对批量请求的延迟影响不小。
import httpx
async def fetch_price(client: httpx.AsyncClient, item_id: int) -> dict:
resp = await client.get(f"https://api.example.com/price/{item_id}", timeout=5.0)
if resp.status_code == 429 or resp.status_code >= 500:
raise ConnectionError(f"item {item_id} 状态码 {resp.status_code}")
resp.raise_for_status()
return resp.json()
外部建一个 AsyncClient 复用，不要每次请求新建。连接池大小可以设成和 semaphore 的并发度一致，避免连接池成为新的瓶颈。429 单独处理成可重试错误，这个在下游有速率限制的时候很关键——重试和限流是两件事，限流是为了不触发 429，重试是为了在偶发 429 之后还能救回来。
几点经验
第一，TaskGroup 的“一个失败全取消”对批量抓取来说，往往不是想要的行为。解决办法不是退回 gather，而是在单条任务内部把可恢复错误吃掉，只让不可恢复错误往上走。这样两种语义都能要。
第二，TaskGroup 里的任务如果抛出了 ExceptionGroup，外层一定要用 except* 处理，否则会以未捕获异常结束，日志里看到的是一堆嵌套的 traceback，不好读。
第三，Semaphore 名字里带 lock，但它是计数器，和 Lock 不一样，允许 N 个任务同时进入。并发度设多少，取决于下游能撑多少，不是取决于 CPU 核数——异步任务大部分时间在等网络，和 CPU 无关。
第四，asyncio.timeout 在派生任务里也有作用域限制，如果你在 TaskGroup 内部又起了一层不带 timeout 的 TaskGroup，外层超时足够强，内层也会被取消，但如果你自己用 asyncio.shield 保护了什么，那些东西不会被取消。这个坑我只踩过一次，那次盾牌保护的是个日志上传任务，结果是主流程超时返回了，日志还在后台上传，对上游来说又慢又乱。
第五，整个方案的版本门槛是 3.11。如果服务还在 3.10，只能用 gather 配合手写取消逻辑。那块代码里最麻烦的不是取消本身，是取消之后要等所有任务真正结束，否则 semaphore 释放顺序和连接回收会出问题。升级到 3.11 之后，这些手工活全部可以被 TaskGroup 接管，代码能少一半。
