Python asyncio 任务组：结构化并发与异常处理

在异步编程中，结构化并发是一种管理并发任务的范式，它确保所有任务被正确组织、启动和清理，避免资源泄漏或未处理异常。Python 的 asyncio 库（Python 3.11 及以上版本）引入了 asyncio.TaskGroup 来实现这一目标。任务组允许您将多个协程任务分组管理，并提供内置的异常处理机制。下面我将逐步解释其原理、使用方法和异常处理策略。

1. 结构化并发的基本概念

结构化并发强调任务的生命周期应被限定在一个显式的作用域内（如 async with 块）。这意味着：

• 所有任务必须在同一个父作用域中启动和结束。
• 如果任何任务失败（抛出异常），整个任务组会自动取消所有未完成的任务，防止“僵尸任务”。
• 这类似于同步编程中的 try-finally 块，但专为并发设计。

asyncio.TaskGroup 是实现这一点的核心工具。它通过上下文管理器来管理任务，确保资源及时释放。

2. 使用 asyncio.TaskGroup 创建任务组

以下是一个简单示例，展示如何创建任务组并运行多个协程任务：

import asyncio

async def task1():
    print("任务1开始")
    await asyncio.sleep(1)
    print("任务1结束")

async def task2():
    print("任务2开始")
    await asyncio.sleep(2)
    print("任务2结束")

async def main():
    async with asyncio.TaskGroup() as tg:  # 进入任务组作用域
        tg.create_task(task1())  # 创建任务1
        tg.create_task(task2())  # 创建任务2
    print("所有任务完成")

asyncio.run(main())

• 关键点：
  • async with asyncio.TaskGroup() as tg: 定义一个任务组作用域。所有任务必须在其中创建。
  • tg.create_task(coroutine): 将协程添加到任务组中，并自动启动。
  • 当作用域结束时（async with 块退出），任务组会等待所有任务完成。如果所有任务正常结束，程序继续执行。

3. 异常处理机制

任务组的强大之处在于其内置的异常处理。如果任何任务抛出异常：

• 任务组会立即取消所有其他运行中的任务。
• 异常被收集并向上抛出，您可以使用 try-except 块捕获。
• 这确保了错误不会静默失败，并简化了清理逻辑。

示例：捕获任务组中的异常

import asyncio

async def task_success():
    print("成功任务开始")
    await asyncio.sleep(1)
    print("成功任务结束")

async def task_fail():
    print("失败任务开始")
    await asyncio.sleep(0.5)
    raise ValueError("模拟错误")  # 故意抛出异常

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_success())
            tg.create_task(task_fail())  # 此任务会失败
    except Exception as e:
        print(f"捕获到异常: {e}")  # 处理异常
    print("程序继续运行")

asyncio.run(main())

• 输出解释：
  • 当 task_fail 抛出 ValueError 时，任务组会取消 task_success（即使它还在运行）。
  • 异常被 try-except 捕获，打印错误信息。
  • 程序不会崩溃，而是继续执行后续代码。
• 最佳实践：
  • 在 try-except 中包裹整个任务组，以处理所有可能的异常。
  • 使用特定异常类型（如 ValueError）来精细控制错误处理。
  • 任务组确保即使有异常，资源（如网络连接）也会被自动清理。

4. 高级用法：嵌套任务组和超时控制

任务组支持嵌套使用，以构建更复杂的并发结构。例如：

async def child_group():
    async with asyncio.TaskGroup() as inner_tg:
        inner_tg.create_task(asyncio.sleep(1))
        inner_tg.create_task(asyncio.sleep(2))

async def main():
    async with asyncio.TaskGroup() as outer_tg:
        outer_tg.create_task(child_group())  # 嵌套任务组
        outer_tg.create_task(asyncio.sleep(0.5))
    print("嵌套任务组完成")

• 如果内层任务组抛出异常，外层任务组也会触发取消机制。
• 添加超时控制：结合 asyncio.timeout() 来限制任务执行时间，避免无限等待。

总结

asyncio.TaskGroup 是处理结构化并发和异常处理的理想工具：

• 优点：简化任务管理、自动取消机制、错误传播清晰。
• 适用场景：Web 服务器并发请求、批量数据处理或任何需要可靠并发控制的场景。
• 注意事项：确保使用 Python 3.11+；避免在任务组外创建任务，以防破坏结构。

通过以上步骤，您可以高效地管理异步任务，并确保异常被稳健处理。如果需要更复杂的示例（如结合数据库操作），请提供具体需求！