先说结论：如果你现在还在用 asyncio.gather 拼那种几十上百个 URL 的批量请求，建议花十分钟把 TaskGroup 换过来。它的语义更接近“要么全成，要么全收”，配合限流和重试，代码意外地短。
下面这份代码是我最近写一个数据采集小工具时用到的完整实现，可以直接跑，也可以拿去改。依赖只有一个 httpx，Python 版本要求 3.11 以上（因为用到了 asyncio.timeout）。
先聊聊 gather 那点别扭的地方
asyncio.gather 最让人难受的，是它在出错之后的收尾方式。默认 return_exceptions=False，第一个任务抛异常，gather 立刻把这个异常甩给调用方，可其余任务并不会被取消——它们继续在后台跑，占着连接、占着内存，直到自己结束。你在日志里看到一条报错，然后进程退出时一堆 “Task was destroyed but it is pending” 的警告，基本就是这个原因。
有人会说，加个 return_exceptions=True 不就完了？那样确实不会中断，但你又得手动去结果列表里挑哪个是异常、哪个是数据，而且取消信号也传不进去，Ctrl+C 的时候照样一地鸡毛。
TaskGroup 把这件事收回来了：组内任一子任务失败，其余子任务会被取消并等待收尾，然后以一个 ExceptionGroup 的形式把错误抛出来。这就是所谓的结构化并发——任务的生死由它所处的作用域决定，作用域没退出，就不会有孤儿任务到处乱跑。
完整实现
先看代码，后面再逐段拆解。保存成 crawl.py 就能运行。
import asyncio
import random
import time
from dataclasses import dataclass
import httpx
DEFAULT_HEADERS = {
"User-Agent": "taskgroup-demo/1.0",
"Accept-Encoding": "gzip, deflate",
}
@dataclass
class FetchResult:
url: str
status: int
attempts: int
size: int
elapsed: float
async def _backoff(attempt: int, base_delay: float) -> None:
"""指数退避 + 抖动，避免重试撞车。"""
delay = base_delay * (2 ** (attempt - 1))
await asyncio.sleep(delay + random.uniform(0, delay * 0.3))
async def fetch_with_retry(
client: httpx.AsyncClient,
url: str,
sem: asyncio.Semaphore,
*,
timeout: float = 5.0,
retries: int = 3,
base_delay: float = 0.3,
) -> FetchResult:
if retries < 1:
raise ValueError("retries 至少为 1")
started = time.perf_counter()
# 名额在整条重试链路外获取，重试期间也占着并发位
async with sem:
for attempt in range(1, retries + 1):
try:
resp = await client.get(url, timeout=timeout)
resp.raise_for_status()
return FetchResult(
url=url,
status=resp.status_code,
attempts=attempt,
size=len(resp.content),
elapsed=time.perf_counter() - started,
)
except httpx.HTTPStatusError as exc:
# 4xx 是客户端自己的问题，重试没有意义
if exc.response.status_code < 500 or attempt == retries:
raise
await _backoff(attempt, base_delay)
except (httpx.TimeoutException, httpx.TransportError):
if attempt == retries:
raise
await _backoff(attempt, base_delay)
raise RuntimeError("unreachable")
async def crawl(
urls: list[str],
*,
concurrency: int = 8,
timeout: float = 5.0,
retries: int = 3,
total_timeout: float = 60.0,
) -> tuple[list[FetchResult], list[tuple[str, str]]]:
sem = asyncio.Semaphore(concurrency)
results: list[FetchResult] = []
errors: list[tuple[str, str]] = []
# 连接池上限跟信号量对齐，否则信号量形同虚设
limits = httpx.Limits(
max_connections=concurrency,
max_keepalive_connections=concurrency,
)
async def worker(url: str) -> None:
try:
item = await fetch_with_retry(
client, url, sem, timeout=timeout, retries=retries
)
results.append(item)
except Exception as exc:
# CancelledError 继承自 BaseException，不会被这里捕获
errors.append((url, f"{type(exc).__name__}: {exc}"))
async with httpx.AsyncClient(headers=DEFAULT_HEADERS, limits=limits) as client:
try:
async with asyncio.timeout(total_timeout):
async with asyncio.TaskGroup() as tg:
for url in urls:
tg.create_task(worker(url))
except TimeoutError:
errors.append(("*", f"整体超时，{total_timeout}s 内未跑完"))
return results, errors
async def main() -> None:
urls = [f"https://httpbin.org/status/200?n={i}" for i in range(40)]
t0 = time.perf_counter()
results, errors = await crawl(urls, concurrency=10, total_timeout=30)
cost = time.perf_counter() - t0
print(f"共 {len(urls)} 条，成功 {len(results)}，失败 {len(errors)}，耗时 {cost:.2f}s")
retried = [r for r in results if r.attempts > 1]
print(f"其中重试过的有 {len(retried)} 条")
for url, reason in errors[:5]:
print(" !", url, reason)
if __name__ == "__main__":
asyncio.run(main())逐段说明
信号量包住整条重试链路
注意 async with sem 的位置：它在 for attempt 循环外面。这一点很关键。如果只把单次 client.get 包在信号量里，那么一旦进入退避等待，名额就被释放了，重试请求会和别的任务挤在一起，实际并发数就不可控了。
代价是：退避期间名额被白白占着。流量小的场景无所谓，如果你要处理几千条 URL 且失败率高，可以改成每个 attempt 单独获取信号量，用吞吐换稳定性。
只对值得重试的错误重试
代码里分了两支。HTTP 状态码错误只重试 5xx，4xx 直接抛出——404、403 这类错误你再重试一百次也是同样的结果，纯属浪费对端资源，还可能触发风控。传输层错误（连接被重置、DNS 抖动、读取超时）则一律重试。
退避用的是 base_delay * 2 ** (attempt - 1)，再叠一个 0 到 30% 的随机抖动。抖动这步不是可选项，几十个任务同时失败同时重试，很容易把对端压出 429。
worker 内部把异常吃掉
这是最容易写错的地方。如果把 fetch_with_retry 直接丢进 tg.create_task()，那么任何一条 URL 最终失败，整个 TaskGroup 就会把剩下的任务全取消——你在跑 500 条数据，其中 1 条 404，剩下 300 条全废了。
所以 worker 里套了一层 try/except，把失败降级成错误记录。TaskGroup 的 fail-fast 语义这时候变成了“内部兜底之后，整体仍然完整跑完”。反过来说，如果某类错误你确实希望中断整批，那就别在 worker 里接，让它往上抛。
两层超时各管各的
代码里有两层超时。client.get(url, timeout=timeout) 这一层是单次请求的，5 秒没响应就换一次重试；外面的 asyncio.timeout(total_timeout) 则是整批任务的预算，30 秒还没跑完就直接放弃，剩下的全部取消。
如果没有外层这个预算，一批 2000 条的 URL 在对方持续慢响应的场景下，你能跑十几分钟都出不来。有了它，最坏情况是有界的。
四个容易踩的坑
- 不要提前创建、重复 await。
tg.create_task()直接接收协程对象就行，不需要先asyncio.ensure_future包一层，也不要在外面再await一次。 - 连接池和信号量必须对齐。
httpx.Limits(max_connections=...)如果比信号量小，信号量根本不起作用——请求会卡在连接池的等待队列里，看起来像“卡死”。反过来设得太大，又会绕过限流把对端打爆。 - 取消信号要能穿透。
asyncio.CancelledError在 3.8 之后就继承自BaseException了，所以except Exception不会误吞取消。但如果你写了except BaseException或者裸except:，Ctrl+C 就废了，进程得等所有任务自己跑完才能停。 - 别和 gather 混着写。在一个 TaskGroup 里面再起 gather，取消的传播路径会变得很难推理。要用就统一用 TaskGroup，嵌套也是可以的，层级清晰。
怎么验证效果
跑完看三个数就够了：总耗时、重试过的条数、失败清单。
假设平均单条延迟 0.3 秒，40 条串行大约 12 秒，并发 10 之后理论值在 1.5 秒上下，实际会略高一些，因为连接建立、TLS 握手和调度都有开销。如果实测明显偏离，通常就是连接池或者信号量的配置没对齐。
重试条数则反映对端的稳定性。如果这个数字长期偏高，说明你该降低并发，而不是加大重试次数——重试是兜底手段，不是性能优化手段。
结尾
这套结构我已经在几个采集脚本里跑了大半年，改动最少的地方恰恰是并发控制那部分——TaskGroup 把任务的边界画清楚了，后面加指标、加日志、加缓存都能塞在 worker 里，不用动骨架。
如果你的项目还在 3.10 及以下，可以用 asyncio.gather(..., return_exceptions=True) 加手动取消来近似模拟，但那种写法在异常路径上永远比 TaskGroup 多几行，而且总有一天会被某个人改坏。
