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 asyncioimport functoolsasync 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 asyncioimport sysdef 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
应用程序，确保即使在并发环境下，异常也能得到妥善的处理和报告。
以上就是Python asyncio 的未处理异常提示的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1372360.html
微信扫一扫 
支付宝扫一扫 
