Python asyncio 3.12 进阶：任务组与结构化并发的最佳实践

在 Python 3.12 中，asyncio 库引入了强大的 TaskGroup 功能，支持结构化并发（Structured Concurrency）范式。结构化并发确保并发任务有明确的生命周期，避免资源泄漏和未处理异常，使代码更健壮、可维护。任务组允许你管理一组相关任务作为一个整体，提供内置的错误传播和取消机制。下面我将逐步解释核心概念、提供代码示例，并总结最佳实践。

1. 核心概念：任务组与结构化并发

• 任务组（Task Group）：通过 asyncio.TaskGroup 类实现，它使用上下文管理器（async with）创建一组任务。当退出上下文时，任务组会等待所有任务完成（或取消），并处理异常。
• 结构化并发（Structured Concurrency）：这是一种编程模型，强调并发操作必须在严格的作用域内开始和结束。所有任务在父任务的作用域内执行，确保：
  • 任务不会“泄漏”（即所有任务都被追踪）。
  • 异常不会静默失败（错误会向上传播）。
  • 资源自动清理（如网络连接或文件句柄）。
• 在 Python 3.12 中，TaskGroup 取代了旧的 gather 或 wait 方法，提供更安全、更直观的并发管理。

2. 代码示例：基本用法与错误处理

以下示例演示如何使用 TaskGroup 创建并发任务，并处理异常。任务组确保所有任务在退出上下文前完成，如果任一任务失败，整个组会取消并传播异常。

import asyncio

async def worker(name: str, delay: float):
    """模拟一个异步任务，可能成功或失败。"""
    await asyncio.sleep(delay)
    print(f"任务 {name} 开始执行")
    if name == "B":  # 模拟任务B失败
        raise RuntimeError(f"任务 {name} 出错")
    print(f"任务 {name} 完成")

async def main():
    """主函数，使用任务组管理并发任务。"""
    try:
        async with asyncio.TaskGroup() as tg:  # 进入任务组上下文
            # 创建多个任务，作为组的一部分
            tg.create_task(worker("A", 1.0))  # 任务A延迟1秒
            tg.create_task(worker("B", 0.5))  # 任务B延迟0.5秒（会失败）
            tg.create_task(worker("C", 1.5))  # 任务C延迟1.5秒
    except* RuntimeError as e:  # 处理任务组中的异常（Python 3.11+ 语法）
        print(f"捕获到任务组异常: {e.exceptions}")
    finally:
        print("所有任务已处理，资源清理完成")

# 运行主函数
asyncio.run(main())

输出解释：

• 任务B 会先执行并抛出异常。
• 任务组会取消所有未完成的任务（如任务C），并捕获异常。
• except* 块处理异常（Python 3.11+ 支持异常组），确保错误不会静默失败。
• finally 块保证资源清理。

3. 最佳实践：结构化并发的关键原则

遵循这些最佳实践，能最大化利用任务组，避免常见陷阱：

1. 始终使用 TaskGroup 管理任务作用域
   不要手动创建 asyncio.create_task() 而不管理。任务组确保所有任务在父作用域内结束，防止“孤儿任务”。示例：

   # 正确做法：使用 TaskGroup 上下文
   async with asyncio.TaskGroup() as tg:
       tg.create_task(task1())
       tg.create_task(task2())
   
   # 错误做法：直接创建任务，可能导致任务泄漏
   asyncio.create_task(task1())  # 无管理，任务可能无限运行
   

2. 正确处理异常
   使用 try-except 包裹任务组上下文，捕获 ExceptionGroup（Python 3.11+）。如果任务失败，任务组会自动取消其他任务，并聚合异常。

   • 在任务内部：使用局部 try-except 处理可恢复错误。
   • 在任务组外部：使用 except* 处理全局错误。

   try:
       async with asyncio.TaskGroup() as tg:
           tg.create_task(worker("A", 1.0))
   except* Exception as e:  # 捕获所有异常组
       print(f"错误: {e}")
   

3. 结合超时机制
   使用 asyncio.timeout() 防止任务永久阻塞。设置全局或任务级超时，确保系统响应性。

   async with asyncio.TaskGroup() as tg:
       # 为单个任务设置超时
       async with asyncio.timeout(2.0):  # 2秒超时
           tg.create_task(worker("A", 3.0))  # 如果超时，会引发 TimeoutError
   

4. 避免嵌套任务组，保持结构扁平
   每个任务组应管理一个逻辑单元的任务。过度嵌套会使代码复杂化。优先使用单一任务组，除非有明确的分层需求。

   # 推荐：单一任务组
   async with asyncio.TaskGroup() as tg:
       tg.create_task(sub_task1())
       tg.create_task(sub_task2())
   
   # 避免：不必要的嵌套
   async with asyncio.TaskGroup() as tg1:
       tg1.create_task(task1())
       async with asyncio.TaskGroup() as tg2:  # 可能增加复杂度
           tg2.create_task(task2())
   

5. 资源清理与取消处理
   在任务中使用 finally 块或 async with 资源管理器（如 aiofiles），确保网络连接、文件句柄等被释放。任务组在取消时，会触发任务的 CancelledError。

   async def worker_with_resource(name):
       try:
           # 模拟资源使用（如数据库连接）
           print(f"任务 {name} 获取资源")
           await asyncio.sleep(1.0)
       finally:
           print(f"任务 {name} 释放资源")  # 即使任务取消，也会执行
   

6. 任务命名与日志追踪
   为任务设置描述性名称，便于调试。使用 asyncio 的日志工具监控任务生命周期。

   async with asyncio.TaskGroup() as tg:
       task = tg.create_task(worker("A", 1.0), name="重要任务")
       print(f"任务名称: {task.get_name()}")  # 输出：重要任务
   

4. 总结

在 Python 3.12 中，asyncio.TaskGroup 是实现结构化并发的最佳工具。它通过严格的作用域管理，提升了代码的可靠性和可维护性。关键优势包括：

• 自动错误传播：异常不会丢失，简化调试。
• 隐式取消：任务失败时，组内其他任务被取消，避免资源浪费。
• 简化代码：减少样板代码（如手动 gather 或 wait）。

通过遵循上述最佳实践，你能构建高效、健壮的异步应用。实际开发中，建议参考 Python 官方文档 asyncio.TaskGroup 进行深入学习。