Python中的协程任务组管理与gather()、wait()方法的区别

知识点描述
在Python异步编程中，asyncio.gather()和asyncio.wait()都是用于并发执行多个协程任务的重要函数，但它们在设计目的、行为特性和使用场景上存在显著差异。理解这两个函数的区别对于编写高效、可靠的异步代码至关重要。

详细讲解

1. 基本概念回顾

• 协程任务（Task）：对协程的封装，表示一个可调度的异步计算单元
• 任务组管理：需要同时启动多个异步操作，并等待它们全部或部分完成
• 并发执行：通过事件循环调度，实现多个任务的交替执行（非真正并行）

2. asyncio.gather() 详解

设计目标：收集多个协程的结果，强调"全部完成"的语义

基本用法：

import asyncio

async def task1():
    await asyncio.sleep(1)
    return "结果1"

async def task2():
    await asyncio.sleep(2)
    return "结果2"

async def main():
    # gather返回结果顺序与传入顺序一致
    results = await asyncio.gather(task1(), task2())
    print(results)  # ['结果1', '结果2']

关键特性：

• 结果顺序保证：返回结果列表与传入协程的顺序完全对应
• 全有或全无：默认情况下，任一任务异常会导致整个gather异常
• 异常处理：可通过return_exceptions=True参数让异常作为结果返回
• 取消传播：取消gather会取消所有未完成的任务

异常处理示例：

async def failing_task():
    raise ValueError("出错了")

async def main():
    # 方式1：默认行为（立即抛出第一个异常）
    try:
        await asyncio.gather(task1(), failing_task())
    except ValueError as e:
        print(f"捕获异常: {e}")
    
    # 方式2：异常作为结果返回
    results = await asyncio.gather(task1(), failing_task(), 
                                  return_exceptions=True)
    print(results)  # ['结果1', ValueError("出错了")]

3. asyncio.wait() 详解

设计目标：提供更精细的任务控制，支持多种完成条件

基本用法：

async def main():
    # 创建任务对象
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    
    # 等待所有任务完成
    done, pending = await asyncio.wait([task1_obj, task2_obj])
    
    # 手动处理结果
    for task in done:
        if task.exception():
            print(f"任务异常: {task.exception()}")
        else:
            print(f"任务结果: {task.result()}")

关键特性：

• 精细控制：返回已完成(done)和未完成(pending)的任务集合
• 多种等待策略：通过return_when参数控制等待条件
• 手动结果处理：需要遍历任务集合获取结果或异常
• 灵活的错误处理：异常不会自动传播，需要手动检查

等待策略示例：

async def main():
    tasks = [asyncio.create_task(task1()), asyncio.create_task(task2())]
    
    # 策略1：第一个任务完成时返回
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print(f"已完成: {len(done)}, 未完成: {len(pending)}")
    
    # 策略2：第一个异常时返回  
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
    
    # 策略3：所有任务完成（默认）
    done, pending = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)

4. 核心差异对比

4.1 设计哲学差异

• gather()：面向结果的批量操作，适合"需要所有结果"的场景
• wait()：面向过程的精细控制，适合需要中间状态监控的场景

4.2 参数传递方式

# gather支持直接传递协程或任务，自动封装
results = await asyncio.gather(coro1(), coro2(), coro3())

# wait必须传递可等待对象（通常是任务对象）
tasks = [asyncio.create_task(coro1()), asyncio.create_task(coro2())]
done, pending = await asyncio.wait(tasks)

4.3 异常处理机制

• gather：异常立即传播或作为结果返回，处理简单统一
• wait：异常封装在任务对象中，需要手动检查每个任务

4.4 超时处理

# gather的超时会影响整个操作
try:
    results = await asyncio.gather(task1(), task2(), timeout=5)
except asyncio.TimeoutError:
    print("整体超时")

# wait的超时只影响等待操作，任务继续运行
done, pending = await asyncio.wait(tasks, timeout=5)
for task in pending:
    task.cancel()  # 通常需要手动取消未完成任务

5. 实际应用场景选择

适合使用gather的场景：

• 需要所有任务的返回结果，且结果顺序重要
• 简单的错误处理需求（要么全部成功，要么整体失败）
• 批量操作，如并发请求多个API并收集结果

适合使用wait的场景：

• 需要监控任务执行进度，如实现进度条
• 需要根据完成情况动态调整，如"完成一个处理一个"
• 需要精细的超时控制或取消策略
• 实现复杂的错误恢复逻辑

6. 性能考虑与最佳实践

资源管理：

# 避免创建过多并发任务
async def process_batch(items, max_concurrency=10):
    semaphore = asyncio.Semaphore(max_concurrency)
    
    async def process_item(item):
        async with semaphore:
            return await process_single_item(item)
    
    # 使用gather控制并发数量
    return await asyncio.gather(*[process_item(item) for item in items])

错误恢复模式：

async def robust_operation(tasks, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await asyncio.gather(*tasks)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 指数退避

总结：gather()提供了简单直观的批量操作接口，而wait()提供了更强大的流程控制能力。选择哪个函数取决于具体的业务需求：如果只是需要并发执行并收集结果，使用gather()；如果需要精细控制执行流程或处理复杂的并发场景，使用wait()。