TaskGroup 是 Python 3.11 引入的结构化并发机制，强制子任务生命周期绑定于上下文，任一未处理异常即取消其余任务；而 asyncio.gather 仅调度并发，不保证任务终止一致性与异常传播可控性。
TaskGroup 是什么，它和 asyncio.gather 有什么本质区别？
TaskGroup 是 Python 3.11 引入的正式化并发控制机制，核心目标是结构化并发（structured concurrency）：所有子任务必须在 TaskGroup 上下文退出前完成或被显式取消，不会出现“幽灵任务”漏跑或失控。而 asyncio.gather 只是并发调度工具，不强制生命周期绑定——比如某个请求抛出未捕获异常，其余任务可能继续运行，也可能静默中断，行为不可预测。
实际用起来，TaskGroup 更适合需要强错误隔离、资源清理明确的自动化请求场景，比如批量调用第三方 API 并统一处理超时/失败/限流响应。
怎么写一个带超时和错误聚合的并发请求函数？
关键不是“并发越多越好”，而是让每个请求有独立超时、可区分错误类型，并在任一失败时快速终止全部（或按需继续）：
import asyncio
import aiohttp
<p>async def fetch_with_timeout(session, url, timeout=5):
try:
async with session.get(url, timeout=timeout) as resp:
return await resp.text()
except asyncio.TimeoutError:
raise TimeoutError(f"Request to {url} timed out")
except aiohttp.ClientError as e:
raise ConnectionError(f"Network error for {url}: {e}")</p><p>async def batch_fetch(urls):
async with aiohttp.ClientSession() as session:
results = {}
errors = {}</p><h1>注意：这里用 TaskGroup 而不是 gather，才能保证异常传播可控</h1><pre class='brush:python;toolbar:false;'> async with asyncio.TaskGroup() as tg:
tasks = {
url: tg.create_task(fetch_with_timeout(session, url))
for url in urls
}
# 所有任务已结束，逐个取结果或异常
for url, task in tasks.items():
if task.done() and not task.cancelled():
try:
results[url] = task.result()
except Exception as e:
errors[url] = e
return results, errors要点：
立即学习“Python免费学习笔记（深入）”；
-
TaskGroup内创建的任务，一旦任一任务因未处理异常退出，整个组会立即取消其余正在运行的任务（除非显式设cancel_remaining=True） - 不要在
TaskGroup外部 await 单个 task —— 它可能已被取消，task.result()会抛CancelledError - 超时必须设在请求层（如
aiohttp的timeout=），而不是靠asyncio.wait_for包裹整个create_task，否则取消逻辑不干净
为什么并发数提不上去？常见瓶颈在哪？
并发请求数上不去，90% 不是 TaskGroup 的问题，而是底层限制没调好：
常见卡点：
-
aiohttp.ClientSession默认连接池上限是 100，但单域名默认只开 10 个并发连接 —— 需显式配置：connector=aiohttp.TCPConnector(limit_per_host=50) - 目标服务端限流（如 GitHub API 每小时 5000 次），再多并发也只会收到 429 响应，得加退避逻辑
- 本地文件描述符不足（Linux 默认常为 1024），大量并发 socket 会报
OSError: [Errno 24] Too many open files，需调高 ulimit 或复用 session -
TaskGroup本身无并发数量限制，但 Python 的 event loop 调度开销在几千 task 级别才开始明显，一般不用刻意拆分批次
TaskGroup 在真实自动化脚本里怎么嵌套使用？
自动化流程往往多层嵌套：比如先并发获取一批 ID 列表，再对每个 ID 并发拉取详情。这时不能简单套两层 TaskGroup，因为外层异常会取消内层，但内层失败不该直接 kill 外层。
正确做法是把内层封装成独立协程，由外层 create_task 启动，并自己处理内层异常：
async def fetch_item_details(item_id):
async with aiohttp.ClientSession() as session:
async with asyncio.TaskGroup() as tg:
# 这里并发拉取该 item 的多个字段接口
tg.create_task(session.get(f"/api/items/{item_id}/price"))
tg.create_task(session.get(f"/api/items/{item_id}/stock"))
# ... 其他子请求
# 内层 TaskGroup 的异常在此处被捕获，不影响外层
<p>async def main():
item_ids = await fetch_all_ids() # 第一层并发
async with asyncio.TaskGroup() as tg:
for item_id in item_ids:
tg.create_task(fetch_item_details(item_id)) # 每个 item 独立生命周期</p>真正容易被忽略的是：TaskGroup 不等于“自动优化”，它只是帮你守住并发边界。请求性能最终取决于网络、服务端响应、连接复用和错误重试策略——这些都得自己补全，TaskGroup 只负责不让它们失控。
