Python 的 asyncio.gather 是异步并发中常用的结果聚合器，适合同时请求多个 API、并行查询数据库、批量下载等需要等待全部结果再统一处理的场景。它接收多个 awaitable，并按输入顺序返回结果。本文围绕它的执行流程、异常与取消、并发限制、常见坑和落地封装展开。
一、基本用法与三个核心对象
asyncio.gather 的基本调用形式如下：- results = await asyncio.gather(task1, task2, task3)
- async def fetch_data():
- return 'data'
- coro = fetch_data() # 协程对象，尚未执行
- task = asyncio.create_task(fetch_data()) # Task 包装并调度执行
- future = asyncio.Future()
- future.set_result('done') # 手动设置结果
二、gather 的生命周期
调用 asyncio.gather(*aws) 后，内部先遍历所有输入参数，对每个参数调用 ensure_future()：协程转为 Task，Future/Task 直接使用。接着检查所有 Future 是否属于同一事件循环，并创建 _GatheringFuture 作为聚合器。gather 还会对相同 Future 去重：同一个 Future 传入两次只会被调度一次，但结果会在返回列表中重复出现。
执行阶段，gather 为每个子任务注册完成回调。回调负责计数已完成任务、根据 return_exceptions 判断异常处理方式，并在所有任务完成后汇总结果。结果顺序严格保持与输入参数一致。例如：- async def slow(n):
- await asyncio.sleep(n)
- return n
- results = await asyncio.gather(slow(1), slow(0.5))
- print(results) # [1, 0.5]
三、异常处理：return_exceptions 的两种行为
默认 return_exceptions=False。任一子任务异常会立即传播到聚合器，其他子任务不会被自动取消，仍可能继续执行。例如：- async def bad():
- raise ValueError('oops')
- try:
- await asyncio.gather(good(), bad())
- except ValueError as e:
- print(f'捕获到异常: {e}') # good() 可能仍在执行
- results = await asyncio.gather(good(), bad(), return_exceptions=True)
- # results 可能是 ['good', ValueError('oops')]
四、取消传播与子任务取消
取消 gather 返回的聚合 Future 时，取消请求会传播到所有子任务。聚合器会等待所有子任务完成或取消，最后以 CancelledError 结束：- async def worker():
- try:
- await asyncio.sleep(10)
- except asyncio.CancelledError:
- print('我被取消了')
- raise
- async def main():
- gather_task = asyncio.create_task(asyncio.gather(worker(), worker()))
- await asyncio.sleep(0.1)
- gather_task.cancel()
- try:
- await gather_task
- except asyncio.CancelledError:
- print('聚合器被取消')
五、并发限制、内存控制与 Semaphore
gather 本身不限制并发量。一次性创建大量任务可能造成内存压力，因此不建议对成千上万的 URL 直接全部 gather。可以分批处理：- BATCH_SIZE = 100
- for i in range(0, len(urls), BATCH_SIZE):
- batch = urls[i:i + BATCH_SIZE]
- await asyncio.gather(*[download(url) for url in batch])
- sem = asyncio.Semaphore(10)
- async def limited_download(url):
- async with sem:
- return await download(url)
- await asyncio.gather(*[limited_download(url) for url in urls])
六、常见问题与 safe_gather 封装
gather 默认不会因为一个任务失败而取消其他任务，若只捕获异常而不处理剩余任务，可能导致资源泄漏或静默失败。一个常见的安全封装是：创建任务列表，捕获异常后取消所有未完成任务，并等待它们结束：- async def safe_gather(*coros):
- tasks = [asyncio.create_task(c) for c in coros]
- try:
- return await asyncio.gather(*tasks)
- except Exception as e:
- for t in tasks:
- t.cancel()
- await asyncio.gather(*tasks, return_exceptions=True)
- raise e
- async def traced(task):
- print(f'开始 {task}')
- try:
- result = await task
- print(f'完成 {task}')
- return result
- except Exception as e:
- print(f'失败 {task}: {e}')
- raise
- await asyncio.gather(traced(task1()), traced(task2()))
七、超时、部分成功与进度反馈
结合 asyncio.wait_for 可以实现整体超时：- try:
- await asyncio.wait_for(
- asyncio.gather(task1(), task2()),
- timeout=5.0
- )
- except asyncio.TimeoutError:
- print('整体操作超时')
- results = await asyncio.gather(*tasks, return_exceptions=True)
- success = [r for r in results if not isinstance(r, Exception)]
- if len(success) >= MIN_REQUIRED:
- process(success)
- done = 0
- async def tracked(coro, total):
- global done
- result = await coro
- done += 1
- print(f'{done}/{total} 完成')
- return result
- tasks = [tracked(task(i), len(urls)) for i in range(len(urls))]
- await asyncio.gather(*tasks)
八、内部实现与并发模式对比
gather 返回的聚合器由 _GatheringFuture 实现。它扩展标准 Future，重写 cancel() 方法以支持级联取消，维护子任务列表和完成状态，并处理异常传播与结果收集。每个子任务完成时会触发回调，回调检查全局状态，决定立即传播异常还是等待其他任务。CPython 实现中还有一些优化：使用 weakref 避免循环引用、为同步完成的任务走快速路径、尽量减少回调函数内存占用。
与其他并发模式相比：
as_completed 更适合流式处理，先完成的任务可以先处理：- for fut in asyncio.as_completed(tasks):
- result = await fut
- process_immediately(result)
- done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
九、增强版 safe_gather 与决策建议
原文给出的实战增强版封装综合了并发控制、全局超时、错误处理和资源清理：- async def safe_gather(*coros, timeout=None, max_concurrent=100):
- sem = asyncio.Semaphore(max_concurrent)
- async def limited(coro):
- async with sem:
- return await coro
- tasks = [asyncio.create_task(limited(coro)) for coro in coros]
- try:
- return await asyncio.wait_for(asyncio.gather(*tasks), timeout=timeout)
- except Exception as e:
- for t in tasks:
- t.cancel()
- await asyncio.gather(*tasks, return_exceptions=True)
- raise e
|
发表于