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