写异步代码绕不开并发，而并发最怕的是某个任务卡住，其他任务傻等。Python 3.11之前，我们对付这种情况要么用 `asyncio.wait` 手动管理任务集，要么用 `asyncio.gather` 配合 `return_exceptions=True`，但取消和异常处理总是要自己写，写着写着就变成一团乱麻。
直到我用了 Python 3.11 起加入的 `asyncio.TaskGroup` 和 3.11 同款的 `asyncio.timeout`，这种憋屈感才消失。这篇文章我从实际爬虫和API调用的场景出发，讲讲怎么用这两个东西把并发任务管理得服服帖帖。
一、以前用gather处理超时，总是差点意思
比如你要同时请求三个网站，如果某个网站卡了十秒，你不想等那么久。以前典型写法是这样：
import asyncio
async def fetch_url(url):
await asyncio.sleep(3)
return f"ok: {url}"
async def main():
tasks = [
asyncio.create_task(fetch_url("a")),
asyncio.create_task(fetch_url("b")),
asyncio.create_task(fetch_url("c")),
]
done, pending = await asyncio.wait(tasks, timeout=2)
for task in pending:
task.cancel()
# 处理done里的结果...
这代码能跑，但问题不少：你得手动区分哪些完成了哪些没完成，还要手动取消pending，更麻烦的是如果某个任务抛异常，`asyncio.wait` 不会自动处理，你得遍历done去取`task.exception()`，否则异常会丢失或变成警告。
二、TaskGroup：让并发任务像一个团队一样协作
TaskGroup 是一个异步上下文管理器，你把任务丢进去，它会自动等待所有任务结束，并且只要有一个任务抛异常，其他还没完成的任务会被自动取消。最关键的是，你可以直接看着代码就知道任务边界在哪。
看个最基础的例子：
import asyncio
async def worker(name, delay):
print(f"{name} 启动")
await asyncio.sleep(delay)
return f"{name} 完成"
async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(worker("A", 2))
task2 = tg.create_task(worker("B", 1))
task3 = tg.create_task(worker("C", 3))
# 走到这里说明所有任务都正常完成了
print("所有任务结束")
print(task1.result())
print(task2.result())
print(task3.result())
asyncio.run(main())
TaskGroup 退出时会等待所有子任务结束。如果某个任务抛异常，TaskGroup 会取消其他还没结束的任务，然后抛出这个异常。这样你不用担心任务挂了其他任务还在傻跑。
而且你可以直接通过`task.result()`拿到返回值，不用再自己维护列表。
三、配合asyncio.timeout：给整个TaskGroup加超时
TaskGroup 只能保证任务结束后统一处理，但没法限制整个团队的总耗时。这时候就需要 `asyncio.timeout` 这个上下文管理器了。它是 3.11 加入的，比老的 `asyncio.wait_for` 更好用，可以直接套在 `async with` 外面。
下面这段代码实现“3秒内没跑完就全部取消”：
import asyncio
async def fetch_data(url):
await asyncio.sleep(2) # 模拟耗时的请求
return f"数据 from {url}"
async def main():
try:
async with asyncio.timeout(3):
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch_data("http://a.com"))
t2 = tg.create_task(fetch_data("http://b.com"))
t3 = tg.create_task(fetch_data("http://c.com"))
except TimeoutError:
print("整体超时了，所有任务已被取消")
else:
print("全部正常完成")
print(t1.result())
print(t2.result())
asyncio.run(main())
这里 `asyncio.timeout` 内部会在超时后抛出 `TimeoutError`，同时它也会发出一个取消信号给内部的任务。TaskGroup收到这个取消信号后，会自动等待所有子任务取消完，最后向外抛出 `TimeoutError`。所以你只需要一个 `except TimeoutError` 就能处理整体的超时和取消，不用手动遍历任务来cancel了。
注意：`asyncio.timeout` 是一个异步上下文管理器，但3.11之前的老项目可能用的是 `async with async_timeout.timeout(3)` 来替代，现在官方库已经内置，不用再装第三方包了。
四、如果只想等部分任务超时，其他任务继续跑？
有一种场景：比如并发向多个服务发起请求，有些服务不重要，即使它们超时了也不想让它们拖垮主流程。这种时候你不应该用TaskGroup，因为它会取消所有任务。你可以用 `asyncio.timeout` 包裹单独的任务，或者用 `asyncio.as_completed` 逐个处理。
但更优雅的写法是：把每个子任务单独用 `asyncio.timeout` 包起来，然后放进TaskGroup里，让不重要的任务自己超时后结束，不影响其他任务。
async def timed_worker(name, delay):
try:
async with asyncio.timeout(2):
await asyncio.sleep(delay)
return f"{name} ok"
except TimeoutError:
return f"{name} timeout but fine"
async def main():
async with asyncio.TaskGroup() as tg:
task_a = tg.create_task(timed_worker("A", 1))
task_b = tg.create_task(timed_worker("B", 5))
task_c = tg.create_task(timed_worker("C", 3))
# 这里所有任务都会正常结束，没有异常
print(task_a.result())
print(task_b.result())
print(task_c.result())
B和C虽然超时了，但它们自己把TimeoutError捕获了，返回了结果，所以TaskGroup认为它们没有异常，不会取消其他任务。这种方式很适合那种“尽力而为”的并发请求。
五、关键点：任务取消时如何安全清理资源
TaskGroup自动取消任务时，会向任务的协程内抛出 `CancelledError`。如果你的协程内部有需要清理的资源（比如关闭连接、释放锁），你可以在 `finally` 中处理。但要注意，在 `asyncio.CancelledError` 被触发时，`finally` 里不能再用 `await`，否则会再次抛出别的异常，而且 `finally` 里如果向上抛 `CancelledError` 之外的东西，会覆盖取消。
一个安全示例：
async def handle_request():
conn = await open_conn()
try:
await conn.query()
finally:
# 注意这里的清理操作尽量不用await
conn.close()
如果你必须用 `await` 来清理（比如异步连接池），那么应当先捕获取消异常，做清理，再重新抛出取消异常：
async def safe_cleanup():
conn = await async_open()
try:
await conn.process()
except asyncio.CancelledError:
await conn.close()
raise
except Exception:
await conn.close()
raise
但这样写起来很麻烦。所以我建议：如果确实需要异步清理，尽量在协程里避免使用需要长事件循环的清理逻辑，或者使用 `asyncio.shield` 保护清理过程。不过日常小任务基本用不到这么精细。
六、再看一个真实场景：批量调用第三方API
假设我有一个用户ID列表，需要调用第三方接口获取每个用户的详细信息。但第三方接口有限流，单次调用不能太频繁，而且整个批量操作不能超过5秒，否则会影响到用户等待。
传统写法需要手动信号量限流 + 超时控制，代码冗长。用TaskGroup加timeout可以这样写：
import asyncio
async def fetch_user_info(user_id, semaphore):
async with semaphore:
# 模拟请求延迟
await asyncio.sleep(0.5)
return {"id": user_id, "name": f"用户{user_id}"}
async def main():
user_ids = [i for i in range(10)]
semaphore = asyncio.Semaphore(3) # 最多同时3个请求
try:
async with asyncio.timeout(5):
async with asyncio.TaskGroup() as tg:
tasks = []
for uid in user_ids:
tasks.append(tg.create_task(fetch_user_info(uid, semaphore)))
except TimeoutError:
print("批量请求超时，取消了所有剩余任务")
return
else:
for task in tasks:
print(task.result())
asyncio.run(main())
这个代码里，TaskGroup保证了如果其中一个请求抛出异常，其他还在限流等待的请求会被取消，不会整个卡死。timeout保证了整体耗时不超过5秒。异常处理也集中在一个地方，比原来写一大堆循环优雅多了。
七、关于CancelledError，你千万别乱吞
TaskGroup在取消任务时，如果任务捕获了`CancelledError`但没有重新抛出，TaskGroup会认为任务没有取消成功，一直等它结束，这可能会导致`async with`卡住。所以如果你在任务里显式捕获`CancelledError`，必须重新抛出，除非你有足够强烈的理由去忽略它。
举个例子：
async def bad_worker():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
# 这里不执行 raise，只打印
print("取消了吗？我不走")
如果你把这种任务放进TaskGroup，然后某个异常触发了取消，TaskGroup会等这个任务自然结束，而它还要等10秒！这会让TaskGroup的整体退出变得异常缓慢。所以一定不要吞掉取消异常。
如果你确实需要在取消时清理，正确的写法是：
async def good_worker():
try:
await asyncio.sleep(10)
finally:
print("清理资源")
因为`finally`不会吞掉取消异常，取消后finally执行完，异常继续传播，任务正常结束。这就够了。
八、老版本怎么办？写一个简单兜底
如果你还在用Python 3.10或更低，没有TaskGroup也没关系，可以用`asyncio.wait`自己模拟一个简化版。不过既然Python 3.11已经是主流，再为了老版本放弃这些舒服的API实在不划算。如果公司项目还锁着3.8，那这篇文章就当给你指个方向，等升级后再用。
但好消息是，从Python 3.12开始，TaskGroup的异常传播机制更加稳定，还支持了`asyncio.timeout`与TaskGroup嵌套时的更清晰行为。所以我强烈建议你现在就升级到3.12或3.13，这些特性好用得不是一星半点。
九、想更顺手？把它封装成一个通用函数
我日常使用中，经常需要“并发执行任务，但指定总超时时间”。我封装了一个小工具：
async def run_with_timeout(coros, timeout, *, return_exceptions=False):
"""
并发执行给定的协程列表，支持整体超时，超时后自动取消未完成任务。
返回一个结果列表。
"""
async def wrap(coro):
return await coro
try:
async with asyncio.timeout(timeout):
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(wrap(coro)) for coro in coros]
except TimeoutError:
if return_exceptions:
return []
raise
if return_exceptions:
return [task.result() if not task.cancelled() else None for task in tasks]
return [task.result() for task in tasks]
不过这个函数有缺陷：如果任务自身抛异常，TaskGroup会取消其他任务并抛出异常，这不符合`return_exceptions=True`的期望。真要完整支持，还得在`wrap`里面捕获异常。但用于简单场景已经足够。你可以根据自己项目需要调整。
十、最终心得
写完这个主题，我心里最深的感受是：异步编程的难点不在于语法，而在于对取消、超时和异常处理的把控。TaskGroup和timeout把这三者统一进了一套简洁的模型里，让我不再需要去记“哪个任务还没取消，哪个任务抛了什么异常”。这已经不仅仅是语法糖，而是编程心智的简化。
在我的爬虫项目里，我重写了一半的并发代码，原来偶发的“卡死不动”现象消失了。最重要的是，代码读起来特别顺畅，别人接手时能一眼看出这个并发块从哪里开始，到哪里结束，什么时候超时，超时怎么办。
如果你也在为管理一堆并发任务而头疼，不妨试试TaskGroup和asyncio.timeout。也许你会跟我一样，有那种“终于等到你”的感觉。
