侧边栏壁纸
博主头像
Euler的博客

  • 累计撰写 158 篇文章
  • 累计创建 20 个标签
  • 累计收到 0 条评论

目 录CONTENT

文章目录
后端技术

【后端】asyncio.gather异常处理与分散-聚集模式详解

EulerBlind
EulerBlind
2025-09-15 / 0 评论 / 0 点赞 / 1 阅读 / 0 字

概述

asyncio.gather是Python异步编程中用于并发执行多个协程的核心函数，它实现了分散-聚集（Scatter-Gather）模式。这种模式在需要同时执行多个异步任务并收集结果的场景中非常有用，但异常处理机制往往让开发者感到困惑。

本文通过实际代码示例深入解析 gather函数的异常处理机制，对比不同参数设置下的行为差异，并提供最佳实践建议。

分散-聚集模式简介

分散-聚集模式是一种并发编程模式，包含两个阶段：

  1. 分散（Scatter）：将任务分发给多个执行单元并发执行
  2. 聚集（Gather）：收集所有执行单元的结果并返回

在异步编程中，这种模式特别适用于：

  • 并行处理多个独立的异步操作
  • 批量数据获取和处理
  • 微服务架构中的聚合操作

基础示例代码

import asyncio
import random
​
async def factorial(name, number, raise_exception=False):
    """
    计算阶乘的异步函数
    
    Args:
        name: 任务名称
        number: 要计算阶乘的数字
        raise_exception: 是否在特定条件下抛出异常
    """
    f = 1
    for i in range(2, number + 1):
        print(f'  Task {name}: Compute factorial({i})...')
        
        # 模拟异步操作
        await asyncio.sleep(0.1)
        
        # 在特定条件下抛出异常
        if raise_exception and i > 3:
            print(f'  Task {name}: raising Exception')
            raise Exception(f'Bad Task {name}')
        
        f *= i
    
    print(f'==>> Task {name} DONE: factorial({number}) = {f}')
    return f
​
async def main():
    """主函数：演示gather的基本用法"""
    tasks = [
        factorial('A', 5),  # 正常任务
        factorial('B', 10, raise_exception=True),  # 会抛出异常的任务
        factorial('C', 2)   # 正常任务
    ]
​
    try:
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print('Results:', results)
    except Exception as e:
        print('Got an exception:', e)
​
if __name__ == "__main__":
    asyncio.run(main())

异常处理机制详解

默认异常处理（return_exceptions=False）

async def main_default():
    """演示默认异常处理行为"""
    tasks = [
        factorial('A', 5),  # 正常任务
        factorial('B', 10, raise_exception=True),  # 会抛出异常
        factorial('C', 2)   # 正常任务
    ]
​
    try:
        # 默认情况下，return_exceptions=False
        results = await asyncio.gather(*tasks)
        print('Results:', results)
    except Exception as e:
        print('Got an exception:', e)
        # 注意：这里无法获取到其他任务的结果

行为特点：

  • 快速失败：当任何一个任务抛出异常时，异常立即传播到主程序
  • 任务中断：异常发生后，其他正在执行的任务会被取消
  • 结果丢失：无法获取已完成任务的结果
  • 程序终止：主程序捕获异常后，整个程序生命周期结束

输出结果：

  Task A: Compute factorial(2)...
  Task B: Compute factorial(2)...
  Task C: Compute factorial(2)...
  Task A: Compute factorial(3)...
  Task B: Compute factorial(3)...
==>> Task C DONE: factorial(2) = 2
  Task A: Compute factorial(4)...
  Task B: Compute factorial(4)...
  Task B: raising Exception
Got an exception: Bad Task B

重要说明：

**虽然任务A在异常发生时被中断，但通过添加 **await asyncio.sleep(10) 可以发现，任务A实际上仍在事件循环中继续执行。这是因为异常只是中断了 gather的等待，但不会自动取消已创建的任务。

添加延迟后的完整输出：

  Task A: Compute factorial(2)...
  Task B: Compute factorial(2)...
  Task C: Compute factorial(2)...
  Task A: Compute factorial(3)...
  Task B: Compute factorial(3)...
==>> Task C DONE: factorial(2) = 2
  Task A: Compute factorial(4)...
  Task B: Compute factorial(4)...
  Task B: raising Exception
Got an exception: Bad Task B
  Task A: Compute factorial(5)...
==>> Task A DONE: factorial(5) = 120

异常作为结果返回（return_exceptions=True）

async def main_with_exceptions():
    """演示异常作为结果返回的行为"""
    tasks = [
        factorial('A', 5),  # 正常任务
        factorial('B', 10, raise_exception=True),  # 会抛出异常
        factorial('C', 2)   # 正常任务
    ]
​
    try:
        # 设置return_exceptions=True，异常将作为结果返回
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print('Results:', results)
        
        # 处理结果，区分正常结果和异常
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f'Task {i} failed with: {result}')
            else:
                print(f'Task {i} succeeded with: {result}')
                
    except Exception as e:
        print('Unexpected exception:', e)

行为特点：

  • 容错执行：异常不会中断其他任务的执行
  • 完整结果：所有任务都会执行完毕，结果按顺序返回
  • 异常封装：异常对象直接包含在结果列表中
  • 灵活处理：可以根据结果类型进行不同的处理逻辑

输出结果：

  Task A: Compute factorial(2)...
  Task B: Compute factorial(2)...
  Task C: Compute factorial(2)...
  Task A: Compute factorial(3)...
  Task B: Compute factorial(3)...
==>> Task C DONE: factorial(2) = 2
  Task A: Compute factorial(4)...
  Task B: Compute factorial(4)...
  Task B: raising Exception
  Task A: Compute factorial(5)...
==>> Task A DONE: factorial(5) = 120
Results: [120, Exception('Bad Task B'), 2]
Task 0 succeeded with: 120
Task 1 failed with: Bad Task B
Task 2 succeeded with: 2

优势对比：

特性 return_exceptions=False return_exceptions=True
异常处理 快速失败，立即中断 容错执行，继续完成
结果获取 无法获取已完成结果 获取所有任务结果
错误恢复 难以实现 容易实现
适用场景 严格依赖所有任务成功 允许部分任务失败

现代替代方案：TaskGroup

asyncio.gather 虽然功能强大，但在异常处理方面存在一些局限性。Python 3.11+ 引入了 asyncio.TaskGroup，提供了更优雅的任务管理和异常处理方式。

TaskGroup 基本用法

async def main_with_taskgroup():
    """使用TaskGroup管理并发任务"""
    results = []
    
    async with asyncio.TaskGroup() as tg:
        # 创建任务并添加到组中
        task_a = tg.create_task(factorial('A', 5))
        task_b = tg.create_task(factorial('B', 10, raise_exception=True))
        task_c = tg.create_task(factorial('C', 2))
        
        # 将任务添加到结果列表
        results.extend([task_a, task_b, task_c])
    
    # 获取所有任务的结果
    print('Results:', [task.result() for task in results])

TaskGroup 的特点：

  • 自动取消：当任何一个任务失败时，所有其他任务会被自动取消
  • 异常传播：第一个异常会立即传播，中断整个组
  • 资源清理：使用 async with 确保资源正确清理
  • 任务管理：提供更好的任务生命周期管理

错误示例：错误的异常处理方式

async def wrong_taskgroup_usage():
    """错误的TaskGroup使用方式"""
    results = []
    
    async with asyncio.TaskGroup() as tg:
        try:
            # 错误：在TaskGroup内部使用try-except无法捕获任务异常
            task_a = tg.create_task(factorial('A', 5))
            task_b = tg.create_task(factorial('B', 10, raise_exception=True))
            task_c = tg.create_task(factorial('C', 2))
            results.extend([task_a, task_b, task_c])
        except Exception as e:
            # 这里的异常处理不会生效
            print('Got an exception:', e)
    
    # 这行代码不会执行，因为TaskGroup会抛出异常
    print('Results:', [task.result() for task in results])

正确的异常处理方式

async def safe_factorial(name, number, raise_exception=False):
    """安全的阶乘计算函数，内部处理异常"""
    try:
        return await factorial(name, number, raise_exception)
    except Exception as e:
        print(f'Task {name} failed: {e}')
        return None
    finally:
        print(f'Task {name} cleanup completed')

async def correct_taskgroup_usage():
    """正确的TaskGroup使用方式"""
    results = []

    async with asyncio.TaskGroup() as tg:
        # 使用内部处理异常的安全函数
        task_a = tg.create_task(safe_factorial('A', 5))
        task_b = tg.create_task(safe_factorial('B', 10, raise_exception=True))
        task_c = tg.create_task(safe_factorial('C', 2))
        results.extend([task_a, task_b, task_c])

    # 获取结果，None表示任务失败
    print('Results:', [task.result() for task in results])

TaskGroup vs gather 对比

特性 asyncio.gather asyncio.TaskGroup
Python版本 3.4+ 3.11+
异常处理 需要手动处理 自动取消其他任务
资源管理 手动管理 自动清理
代码复杂度 相对简单 稍复杂但更安全
适用场景 简单并发任务 复杂任务组管理

实际应用场景

使用 gather 的场景：

  • 简单的批量操作
  • 需要容错处理的场景
  • 对Python版本有要求（< 3.11）

使用 TaskGroup 的场景：

  • 复杂的任务依赖关系
  • 需要严格的任务管理
  • 现代Python环境（3.11+）
  • 需要自动资源清理的场景
0
后端
版权归属： EulerBlind
本文链接： https://blog.nekomint.top/archives/ry1CakXD
许可协议： 本文使用《署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)》协议授权

评论区