Python脚本如何捕获异步任务异常

本文目录导读：

1. 使用 try/except 包裹单个协程
2. 使用 asyncio.gather() 捕获多个任务的异常
3. 使用 asyncio.wait() 分组处理
4. 使用 asyncio.TaskGroup（Python 3.11+）
5. 使用回调函数添加异常处理
6. 全局异常处理器
7. 最佳实践建议

在Python中捕获异步任务的异常有几种常见方法,我来详细介绍每种方式：

使用 try/except 包裹单个协程

import asyncio
async def risky_task():
    await asyncio.sleep(1)
    raise ValueError("出错了！")
async def main():
    try:
        await risky_task()
    except ValueError as e:
        print(f"捕获到异常: {e}")
asyncio.run(main())

使用 asyncio.gather() 捕获多个任务的异常

import asyncio
async def task1():
    await asyncio.sleep(1)
    return "task1完成"
async def task2():
    await asyncio.sleep(2)
    raise RuntimeError("task2失败了")
async def task3():
    await asyncio.sleep(0.5)
    return "task3完成"
async def main():
    try:
        # return_exceptions=False（默认）：第一个异常会立即抛出
        results = await asyncio.gather(task1(), task2(), task3())
        print(f"所有任务完成: {results}")
    except RuntimeError as e:
        print(f"捕获到异常: {e}")
    # 使用 return_exceptions=True 可以获取所有结果和异常
    results = await asyncio.gather(
        task1(), task2(), task3(),
        return_exceptions=True
    )
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务{i+1}失败: {result}")
        else:
            print(f"任务{i+1}成功: {result}")
asyncio.run(main())

使用 asyncio.wait() 分组处理

import asyncio
async def good_task():
    await asyncio.sleep(1)
    return "成功"
async def bad_task():
    await asyncio.sleep(2)
    raise ValueError("失败了")
async def main():
    tasks = [good_task(), bad_task()]
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    for task in done:
        try:
            result = task.result()
            print(f"任务成功: {result}")
        except Exception as e:
            print(f"任务失败: {e}")
    # 取消仍在等待的任务
    for task in pending:
        task.cancel()
asyncio.run(main())

使用 asyncio.TaskGroup（Python 3.11+）

import asyncio
async def task1():
    await asyncio.sleep(1)
    return "task1完成"
async def task2():
    await asyncio.sleep(2)
    raise RuntimeError("task2失败")
async def task3():
    await asyncio.sleep(0.5)
    return "task3完成"
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(task1())
            t2 = tg.create_task(task2())
            t3 = tg.create_task(task3())
        # 如果TaskGroup中没有异常，这里会执行
        print("所有任务完成")
        print(f"task1结果: {t1.result()}")
        print(f"task3结果: {t3.result()}")
    except* RuntimeError as e:
        print(f"捕获到RuntimeError: {e}")
    except* Exception as e:
        print(f"捕获到其他异常: {e}")
asyncio.run(main())

使用回调函数添加异常处理

import asyncio
async def risky_task():
    await asyncio.sleep(1)
    raise TypeError("类型错误")
def handle_exception(future):
    try:
        future.result()
    except Exception as e:
        print(f"回调中捕获异常: {e}")
async def main():
    task = asyncio.create_task(risky_task())
    task.add_done_callback(handle_exception)
    # 等待任务完成
    try:
        await task
    except Exception as e:
        print(f"主协程中也捕获到: {e}")
asyncio.run(main())

全局异常处理器

import asyncio
import sys
def global_exception_handler(loop, context):
    exception = context.get('exception')
    message = context.get('message', '未知错误')
    if exception:
        print(f"全局异常处理: {message}, Exception: {exception}")
    else:
        print(f"全局异常处理: {message}")
async def bad_task():
    await asyncio.sleep(1)
    raise ValueError("任务失败")
async def main():
    # 设置全局异常处理器
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(global_exception_handler)
    # 创建任务但不等待它（模拟未处理的异常）
    asyncio.create_task(bad_task())
    await asyncio.sleep(2)
asyncio.run(main())

最佳实践建议

1. 对于简单场景：使用 try/except 包裹单个 await 调用

2. 多个并行任务：使用 asyncio.gather(return_exceptions=True) 获取所有结果

3. Python 3.11+：优先使用 asyncio.TaskGroup，提供更好的异常处理

4. 需要精细控制：使用 asyncio.wait() 配合 FIRST_EXCEPTION 等选项

5. 复杂应用：设置全局异常处理器来捕获未处理的异常

选择合适的异常处理方式取决于你的具体需求和应用场景。