Python asyncio 的未处理异常提示-Python教程-PHP中文网

Python asyncio中未处理的异常不会立即崩溃程序，而是以警告形式输出，需主动捕获。推荐在协程内用try...except处理异常，或为Task添加done_callback检查结果。使用asyncio.gather(..., return_exceptions=True)可收集多个任务异常而不中断执行。因asyncio任务独立运行，未被捕获的异常会存储于Task对象并最终触发警告，避免单个任务失败导致整个应用崩溃。为确保异常不被遗漏，可设置loop.set_exception_handler()作为全局兜底，但应优先在局部处理异常，避免依赖全局机制。常见陷阱包括忽略await任务、未检查result、误解gather行为及忽视CancelledError处理，正确做法是始终关注任务状态，及时处理异常或取消情况，提升程序健壮性。

python asyncio 的未处理异常提示

Python

asyncio

登录后复制

中未处理的异常，通常不会直接导致程序崩溃，而是以警告的形式打印出来。这看似温柔，实则可能隐藏着更深层次的问题，让开发者难以追踪和调试。核心观点是，你需要主动地、明确地去捕获和处理

asyncio

登录后复制

任务中的异常，而不是寄希望于它会自动中断程序。

解决方案

处理

asyncio

登录后复制

任务中的未处理异常，核心思路是确保每个被创建的任务都有一个“监护人”来关注它的执行结果，特别是异常情况。最直接且推荐的方式是在协程内部使用

try...except

登录后复制

块，这和同步代码中的异常处理并无二致。但对于那些“后台运行”或“fire-and-forget”的任务，你需要一些额外的机制。

一种常见且有效的方法是为任务添加

done_callback

登录后复制

。当一个

asyncio.Task

登录后复制

完成（无论是成功、取消还是异常），这个回调函数都会被调用。你可以在回调中检查任务是否发生了异常。

立即学习“Python免费学习笔记（深入）”；

import asyncio
import functools

async def faulty_coroutine(name):
    print(f"Task {name}: Starting...")
    await asyncio.sleep(0.1)
    if name == "Task B":
        raise ValueError(f"Oops! An error in {name}")
    print(f"Task {name}: Finished successfully.")

def handle_task_exception(task, task_name):
    try:
        task.result() # 尝试获取结果，如果任务有异常，这里会重新抛出
    except asyncio.CancelledError:
        print(f"Task {task_name} was cancelled.")
    except Exception as e:
        print(f"ERROR: Task {task_name} failed with exception: {e}")
        # 这里可以加入日志记录、告警等处理
    else:
        print(f"Task {task_name} completed without exceptions.")

async def main():
    print("Main: Creating tasks...")
    task_a = asyncio.create_task(faulty_coroutine("Task A"))
    task_b = asyncio.create_task(faulty_coroutine("Task B"))
    task_c = asyncio.create_task(faulty_coroutine("Task C"))

    # 为每个任务添加一个回调
    task_a.add_done_callback(functools.partial(handle_task_exception, task_name="Task A"))
    task_b.add_done_callback(functools.partial(handle_task_exception, task_name="Task B"))
    task_c.add_done_callback(functools.partial(handle_task_exception, task_name="Task C"))

    # 等待所有任务完成，但这里不会捕获到 task_b 的异常，因为它已经在回调中处理了
    # 如果不加回调，task_b 的异常会作为警告打印
    await asyncio.gather(task_a, task_b, task_c, return_exceptions=True) # return_exceptions=True 会让 gather 返回异常而非直接抛出

    print("Main: All tasks finished.")

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

登录后复制

在上面的

main

登录后复制

函数中，

asyncio.gather(..., return_exceptions=True)

登录后复制

也是一种捕获多个任务异常的有效方式。当

return_exceptions

登录后复制

设置为

True

登录后复制

时，即使有任务抛出异常，

gather

登录后复制

也不会中断，而是将异常对象作为结果返回。这在处理一组相互独立的任务时非常有用。

为什么

asyncio

登录后复制

任务的异常有时只显示警告而非直接中断程序？

这其实是

asyncio

登录后复制

设计哲学的一部分，尤其是在

asyncio.create_task()

登录后复制

创建的“独立”任务中表现明显。当你通过

asyncio.create_task()

登录后复制

启动一个协程时，你创建了一个

Task

登录后复制

对象，它在事件循环中独立运行。这个任务的生命周期与创建它的父协程在某种程度上是解耦的。如果任务内部发生了未捕获的异常，

asyncio

登录后复制

默认的行为是：它不会立即停止整个事件循环或父协程，而是将异常存储在

Task

登录后复制

对象内部，并在任务完成时（或被垃圾回收时）打印一个警告日志。

这种行为的背后逻辑是，

asyncio

登录后复制

旨在构建高并发、高可用性的应用。一个独立的后台任务的失败，不应该轻易地导致整个应用程序的崩溃，尤其是在服务器应用中。想象一下，如果一个处理用户请求的协程因为某个小错误而崩溃，导致整个服务器进程停止，那将是灾难性的。因此，

asyncio

登录后复制

选择了一种更“宽容”的错误处理方式，它将异常视为任务自身的内部状态，并通过日志警告来通知开发者，而不是强制中断。

然而，这种“宽容”也带来了一定的开发挑战。开发者必须主动地去检查任务的完成状态和潜在的异常，否则这些错误可能会被默默地吞噬，直到在生产环境中引发难以追踪的偶发问题。这要求我们在设计

asyncio

登录后复制

应用时，必须对任务的生命周期和异常处理有清晰的规划。

如何为

asyncio

登录后复制

事件循环设置全局异常处理器？

虽然在任务内部或通过

done_callback

登录后复制

处理异常是推荐的做法，但在某些情况下，你可能希望有一个全局的“最后一道防线”，来捕获那些被遗漏的、未被特定任务处理的异常。

asyncio

登录后复制

提供了

loop.set_exception_handler()

登录后复制

方法，允许你为整个事件循环设置一个自定义的全局异常处理器。

这个全局处理器会在以下几种情况被调用：

一个
```
asyncio.Task
```
登录后复制
抛出了未被
```
await task
```
登录后复制
或
```
task.result()
```
登录后复制
捕获的异常。
```
asyncio
```
登录后复制
内部的一些操作，如
```
call_soon
```
登录后复制
或
```
call_later
```
登录后复制
调用的回调函数抛出了异常。

设置全局异常处理器的代码示例如下：

芦笋演示

一键出成片的录屏演示软件，专为制作产品演示、教学课程和使用教程而设计。

查看详情

import asyncio
import sys

def custom_exception_handler(loop, context):
    exception = context.get("exception")
    message = context.get("message")
    # 打印一些上下文信息，比如任务、协程、堆栈等
    print(f"\n--- Global Exception Handler Caught an Error ---")
    print(f"Message: {message}")
    if exception:
        print(f"Exception Type: {type(exception).__name__}")
        print(f"Exception Value: {exception}")
    print(f"Context: {context}")
    print("--------------------------------------------------")

    # 根据需要，你可以在这里执行一些清理工作，或者决定是否终止程序
    # 例如，如果异常非常严重，可以考虑 sys.exit(1)
    # 但通常，全局处理器更多用于日志记录和告警，而不是直接终止
    # sys.exit(1) # 谨慎使用！

async def another_faulty_coroutine():
    print("Another faulty coroutine running...")
    await asyncio.sleep(0.05)
    raise RuntimeError("This is a runtime error from another coroutine!")

async def main_with_global_handler():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(custom_exception_handler)

    print("Main with global handler: Creating tasks...")
    # 这个任务的异常会被全局处理器捕获
    asyncio.create_task(another_faulty_coroutine())

    # 等待一段时间，让任务有机会抛出异常
    await asyncio.sleep(0.2)
    print("Main with global handler: Finished.")

if __name__ == "__main__":
    try:
        asyncio.run(main_with_global_handler())
    except RuntimeError as e:
        print(f"Caught a RuntimeError outside asyncio.run: {e}")
    except Exception as e:
        print(f"Caught an unexpected error outside asyncio.run: {e}")

登录后复制

通过

loop.set_exception_handler()

登录后复制

，你可以实现一个统一的异常日志记录机制，将所有未捕获的

asyncio

登录后复制

异常汇集到一处处理。这对于生产环境的监控和告警系统集成非常有价值。然而，过度依赖全局处理器也可能导致问题，因为它会掩盖特定任务异常的上下文，让你难以判断异常发生的具体位置和原因。所以，最佳实践仍然是尽可能在异常发生的源头附近进行处理。

在

asyncio

登录后复制

编程中，如何避免常见的异常处理陷阱？

asyncio

登录后复制

的异步特性引入了一些独特的异常处理挑战。要避免常见的陷阱，我们需要对

asyncio

登录后复制

的工作原理有更深入的理解：

忘记

await

登录后复制

任务或检查其结果： 这是最常见的陷阱之一。当你通过

asyncio.create_task(coro())

登录后复制

启动一个任务后，如果你不

await

登录后复制

这个任务，或者不显式地调用

task.result()

登录后复制

或

task.exception()

登录后复制

，那么任务内部抛出的任何异常都可能只以警告形式出现，而不会中断你的主程序流。你必须主动地“关心”这些任务的命运。

陷阱示例：

async def my_bad_task():
    await asyncio.sleep(0.1)
    raise ValueError("I failed!")

async def main_trap():
    asyncio.create_task(my_bad_task()) # 任务启动了，但没人管它的结果
    await asyncio.sleep(0.2) # 主程序继续运行，my_bad_task的异常可能只打印警告

登录后复制

正确做法： 始终

await

登录后复制

你关心的任务，或者为它们添加

done_callback

登录后复制

。

async def main_correct():
    task = asyncio.create_task(my_bad_task())
    try:
        await task # 这里会重新抛出ValueError
    except ValueError as e:
        print(f"Successfully caught error from task: {e}")

登录后复制

过度依赖全局异常处理器： 虽然全局处理器很有用，但它不应该成为你主要的异常处理策略。如果所有的异常都涌向全局处理器，你将失去异常发生的具体上下文信息，导致调试困难。它更适合作为最后的兜底机制，用于捕获那些真正意料之外的、未被局部处理的错误。
对
```
asyncio.gather
```
登录后复制
的
return_exceptions
登录后复制
参数理解不足：
```
return_exceptions=True
```
登录后复制
确实能让
```
gather
```
登录后复制
在有任务失败时继续执行并返回异常，但如果你期望的是任何一个任务失败就立即中断整个
```
gather
```
登录后复制
组，那么你应该省略这个参数（默认为
```
False
```
登录后复制
）。理解其行为，才能正确地设计并发任务的容错逻辑。
在
```
asyncio.run()
```
登录后复制
外部捕获
asyncio
登录后复制
内部异常：
```
asyncio.run()
```
登录后复制
会启动事件循环并运行你的主协程，它内部的异常处理机制是独立的。如果你期望在
```
asyncio.run()
```
登录后复制
的调用点外部捕获到
```
asyncio
```
登录后复制
内部的某个特定异常，那通常是不行的，因为
```
asyncio
```
登录后复制
已经处理（或警告）了。只有当
```
asyncio.run()
```
登录后复制
本身因为某种原因（比如事件循环被关闭）而抛出异常时，外部的
```
try...except
```
登录后复制
才能捕获。
不处理任务取消：
```
asyncio.CancelledError
```
登录后复制
是一个特殊的异常，它表示任务被取消了。如果你在协程内部没有正确处理
```
CancelledError
```
登录后复制
，可能会导致资源泄露或状态不一致。在
```
except
```
登录后复制
块中捕获
```
Exception
```
登录后复制
时，通常也需要单独处理
```
CancelledError
```
登录后复制
。