面试题：Python异步编程的异常处理与最佳实践

使用try - except块捕获异常：
- 在async函数内部，通过try - except块来捕获异步函数执行过程中抛出的异常。
- 例如，如果有一个异步函数async_function可能抛出异常，在调用它的async函数中：
```
async def main():
    try:
        result = await async_function()
    except SomeException as e:
        print(f"捕获到异常: {e}")
```

结合asyncio.gather处理异常：

asyncio.gather用于并发运行多个异步任务。如果其中一个任务抛出异常，默认情况下asyncio.gather不会自动捕获异常，而是让异常传播。
可以通过设置return_exceptions=True参数，使asyncio.gather将异常作为结果返回，而不是直接抛出。这样可以在外部统一处理异常。
示例代码如下：

import asyncio


async def async_task1():
    await asyncio.sleep(1)
    raise ValueError("任务1抛出异常")


async def async_task2():
    await asyncio.sleep(2)
    return "任务2执行成功"


async def main():
    tasks = [async_task1(), async_task2()]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            print(f"捕获到异常: {result}")
        else:
            print(f"任务结果: {result}")


if __name__ == "__main__":
    asyncio.run(main())

自定义异常处理机制：

定义自定义异常类，以便在特定业务场景下抛出和处理异常。
例如：

class MyCustomException(Exception):
    pass


async def async_task_with_custom_exception():
    await asyncio.sleep(1)
    raise MyCustomException("自定义异常发生")


async def main_with_custom_exception():
    try:
        await async_task_with_custom_exception()
    except MyCustomException as e:
        print(f"捕获到自定义异常: {e}")


if __name__ == "__main__":
    asyncio.run(main_with_custom_exception())

复杂异步任务场景下的完整代码示例：

import asyncio


class TaskError(Exception):
    pass


async def async_sub_task1():
    await asyncio.sleep(1)
    raise TaskError("子任务1发生错误")


async def async_sub_task2():
    await asyncio.sleep(2)
    return "子任务2执行成功"


async def async_sub_task3():
    await asyncio.sleep(3)
    return "子任务3执行成功"


async def complex_task():
    sub_tasks = [async_sub_task1(), async_sub_task2(), async_sub_task3()]
    results = await asyncio.gather(*sub_tasks, return_exceptions=True)
    for result in results:
        if isinstance(result, Exception):
            print(f"捕获到异常: {result}")
        else:
            print(f"任务结果: {result}")


if __name__ == "__main__":
    asyncio.run(complex_task())

在上述代码中：

定义了自定义异常类TaskError。
async_sub_task1模拟一个会抛出异常的异步子任务。
async_sub_task2和async_sub_task3是正常执行的异步子任务。
complex_task函数中使用asyncio.gather并发运行这些子任务，并通过检查结果是否为异常类型来处理异常，确保整个异步任务流的稳定性和可靠性。

星途面试题库

面试题：Python异步编程的异常处理与最佳实践

知识考点

面试题答案