用 asyncio.TaskGroup 写出可收敛的并发任务
asyncio.TaskGroup 不是 Python 3.13 才出现的新接口,它从 Python 3.11 起就提供结构化并发。在 Python 3.12 或 3.13 服务中,它可以替代“创建一批任务后希望记得收尾”的做法。核心保证很简单:退出任务组之前,组里的子任务都已经结束;其中一个任务以普通异常失败时,其他任务会被取消,错误再以异常组的形式交给调用者。
生命周期比启动方式更重要
假设一个页面需要并发读取资料、库存和权限。裸用 create_task() 容易丢失任务引用,也容易在请求已结束后继续工作。任务组把创建位置和等待边界放在一起:
import asyncio
async def load_dashboard(user_id: str) -> dict[str, object]:
async with asyncio.TaskGroup() as group:
profile = group.create_task(fetch_profile(user_id))
stock = group.create_task(fetch_stock(user_id))
access = group.create_task(fetch_access(user_id))
return {
"profile": profile.result(),
"stock": stock.result(),
"access": access.result(),
}
只有正常离开 async with 后才读取结果,此时三个任务都已完成。如果库存请求失败,资料与权限任务会收到取消,而调用者不会拿到一半完成、另一半仍在后台运行的模糊状态。这不是数据库事务:已经发送的邮件或已经写入的外部系统不会自动回滚。因此并发组里的副作用仍需幂等键、补偿逻辑或先计算后提交的两阶段设计。
不要吞掉取消
取消通过 asyncio.CancelledError 进入协程。清理资源可以使用 try/finally,但清理完通常应让取消继续传播。如果宽泛地捕获 BaseException 或捕获取消后返回普通结果,任务组的内部取消协议可能被破坏,超时也会表现得像成功。
async def fetch_profile(user_id: str) -> dict:
connection = await open_connection()
try:
return await connection.get(f"/users/{user_id}")
finally:
await connection.close()
这里无需捕获取消;finally 在成功、失败和取消时都会运行。若底层库要求捕获取消做额外记录,记录后应再次 raise。同样,不要把任务组传到生命周期不清楚的全局对象,让后来的代码随意添加任务。
按异常类型处理批量失败
多个子任务可能在取消生效前几乎同时失败。任务组退出时会把非取消异常组合起来。调用者可以用 except* 针对可恢复类型处理,而不是只看“第一个”异常:
try:
result = await load_dashboard("u-42")
except* TimeoutError as failures:
record_timeouts(failures.exceptions)
except* PermissionError:
raise DashboardUnavailable("access denied")
处理策略应靠近能够作决定的层。底层函数保留具体错误和上下文,服务边界再决定重试、降级或向客户端返回什么。不要把所有异常提前压成字符串,否则异常组的结构和 traceback 都会丢失。
超时属于调用范围
asyncio.timeout() 可以包住整个任务组,表达“这个组合最多允许多久”,也可以只包住某个慢依赖。整体超时适合请求预算,局部超时适合允许降级的可选数据。二者同时使用时,应给外层保留清理时间,避免内外预算完全相等而让日志、连接关闭也被截断。
重试不应在每一层重复。为幂等读取设置有限次数、退避和随机抖动;对非幂等写入,必须先有请求标识和服务器端去重。任务组能保证子任务收敛,却不会替你判断重试是否安全。
用确定性测试验证失败路径
并发测试不应依赖真实睡眠和执行顺序。用事件或可控 fake 让一个任务等待,另一个任务抛错,然后断言等待者收到取消且资源完成清理。还要覆盖调用者自身被取消、多个错误形成 ExceptionGroup、整体超时以及成功路径。测试的重点是最终状态和责任边界,不是某次调度恰好先打印哪一行。
结论
TaskGroup 的收益不是少写几行 gather(),而是让父任务对所有子任务负责。把任务创建限制在清晰范围内,保持取消传播,按异常类型处理失败,并把外部副作用设计成可重试或可补偿,才能得到真正“可收敛”的异步代码。结构化并发不会消除失败,但会让失败在调用栈中有确定的归属。