1.协程中的未处理异常会“消失”是因为它们被封装在task对象内或冒泡至事件循环而未被主动检查。2.捕获异常的直接方式是await协程并使用try...except，异常会像同步代码一样传播。3.对于未被await的任务，可通过检查task对象的exception()方法获取异常。4.更优雅的方案是使用task.add_done_callback()添加回调函数，在任务完成时检查异常。5.设置全局事件循环异常处理器是最关键手段，可捕获所有未处理异常，推荐配置以实现统一日志、告警、降级等处理。6.asyncio.gather配合return_exceptions=true可同时运行多个协程并收集结果与异常。7.全局处理器是异步应用的必备实践，提供集中式异常监控，防止错误被忽略，保障系统稳定性。
协程中的未处理异常，说白了，就是那些你以为它会炸出来，结果却悄无声息的“消失”了的错误。这事儿在Python的asyncio里可太常见了，尤其对于刚接触异步编程的朋友来说，简直是噩梦。要发现它们，核心在于理解异步任务的生命周期以及事件循环是如何处理这些“孤儿”错误的。最直接的办法，通常是依赖asyncio提供的异常处理机制，尤其是全局的事件循环异常处理器，它能帮你捕获那些没有被await或try...except捕获的“漏网之鱼”。
要系统性地发现Python协程中未处理的异常，我们需要从几个层面着手。首先，最理想的情况是，你总能await你的协程，这样异常就会像同步代码一样传播开来，你可以用传统的try...except来捕获。但现实往往不是这样，很多协程会被asyncio.create_task()创建后就“放飞自我”了。
对于这些被创建但没有被直接await的任务，它们的异常不会立即中断你的主程序流。这时，你可以通过以下几种方式来“找到”它们：
立即学习“Python免费学习笔记（深入）”；
检查Task对象： 每个由asyncio.create_task()或loop.create_task()创建的协程，都会被封装成一个Task对象。如果这个任务执行过程中发生了异常，并且这个异常没有在任务内部被处理，那么这个异常会被存储在Task对象内部。你可以通过task.exception()方法来获取这个异常（如果任务已完成且有异常）。如果任务正常完成，task.result()会返回结果。
import asyncio
async def faulty_coroutine():
print("Coroutine started, about to raise error...")
raise ValueError("Something went wrong in the coroutine!")
await asyncio.sleep(0.1) # This line won't be reached
async def main_check_task():
task = asyncio.create_task(faulty_coroutine())
# 在实际应用中，你可能需要等待一段时间，确保任务有机会执行
await asyncio.sleep(0.5) # 给任务一些时间运行和失败
if task.done():
if task.exception():
print(f"Task '{task.get_name()}' finished with exception: {task.exception()}")
else:
print(f"Task '{task.get_name()}' finished successfully.")
else:
print(f"Task '{task.get_name()}' is still running.")
# asyncio.run(main_check_task())利用Task.add_done_callback()： 这是更优雅的方案。你可以给一个任务添加一个回调函数，当任务完成（无论是成功、失败还是被取消）时，这个回调函数就会被调用。在回调函数里，你就可以检查任务的异常。
import asyncio
async def another_faulty_coroutine(name):
print(f"Coroutine {name} started, about to raise error...")
await asyncio.sleep(0.1)
if name == "TaskB":
raise TypeError(f"Error from {name}!")
print(f"Coroutine {name} finished successfully.")
def task_done_callback(task):
if task.exception():
print(f"Callback: Task '{task.get_name()}' caught exception: {task.exception()}")
else:
print(f"Callback: Task '{task.get_name()}' completed successfully with result: {task.result()}")
async def main_with_callbacks():
task_a = asyncio.create_task(another_faulty_coroutine("TaskA"), name="TaskA_Coro")
task_b = asyncio.create_task(another_faulty_coroutine("TaskB"), name="TaskB_Coro")
task_a.add_done_callback(task_done_callback)
task_b.add_done_callback(task_done_callback)
# 等待所有任务完成，以便回调函数有机会被调用
await asyncio.gather(task_a, task_b, return_exceptions=True) # return_exceptions=True 确保 gather 不会因为一个任务失败而中断
print("All tasks processed.")
# asyncio.run(main_with_callbacks())设置全局事件循环异常处理器： 这是最关键也是最强大的手段，尤其对于那些你根本没有机会await或添加回调的任务。当一个协程内部的异常没有被任何try...except捕获，也没有被其父任务await，最终它会冒泡到事件循环层。asyncio允许你为事件循环设置一个全局的异常处理器，这样所有这类“未被处理”的异常都会通过这个处理器被捕获。
import asyncio
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
async def very_faulty_coroutine():
print("Very faulty coroutine running...")
await asyncio.sleep(0.1)
raise RuntimeError("This error should be caught by the global handler!")
def custom_exception_handler(loop, context):
exception = context.get("exception")
message = context.get("message", "Unhandled exception in event loop callback")
task = context.get("task")
logging.error(f"Caught unhandled exception in event loop: {message}")
if exception:
logging.error(f"Exception type: {type(exception).__name__}, Value: {exception}")
if task:
logging.error(f"Task involved: {task.get_name() if task else 'N/A'}")
# 这里可以进行更复杂的处理，比如发送报警、记录到日志系统等
# 甚至可以决定是否关闭循环或取消其他任务
# 默认情况下，asyncio会打印错误信息到sys.stderr，我们这里可以自定义行为
# 如果不希望默认行为发生，可以在这里阻止它，但通常建议至少记录下来
async def main_with_global_handler():
loop = asyncio.get_running_loop()
loop.set_exception_handler(custom_exception_handler)
# 创建一个任务，但不await它，也不添加done_callback
# 它的异常最终会冒泡到事件循环，被我们设置的handler捕获
asyncio.create_task(very_faulty_coroutine(), name="Uncaught_Coro")
print("Main function running, waiting for potential exceptions...")
await asyncio.sleep(0.5) # 给任务一点时间执行
# asyncio.run(main_with_global_handler())这个全局处理器是发现真正“未处理”异常的利器。
这个问题，我个人觉得是异步编程最让人迷惑的地方之一。在同步代码里，一个函数抛出异常，如果没被捕获，那程序就直接中断了，非常直观。但在协程里，情况就复杂得多。
想象一下，你启动了一个协程，比如task = asyncio.create_task(my_coroutine())。这个操作只是告诉事件循环：“嘿，我这里有个协程，你找个时间把它跑起来。”事件循环拿到这个指令后，它并不会立即执行my_coroutine()并等待其结果。它只是把my_coroutine()包装成一个Task对象，然后把这个Task扔进它的调度队列里。
接下来，事件循环会继续执行你的主程序代码，或者去跑其他已经准备好的任务。my_coroutine()会在某个未来的时间点被调度执行。如果它在执行过程中抛出了一个异常，而这个异常在my_coroutine()内部没有被try...except捕获，那么这个异常就会被Task对象内部捕获并存储起来。
问题就在于，你的主程序并没有直接await这个Task。所以，主程序并不知道这个Task内部发生了什么。事件循环也不会因为一个非被await的Task内部出现异常就立刻中断整个程序。它会继续运行，就像什么都没发生一样。这就造成了异常“消失”的假象。它不是真的消失了，而是被“封装”在了Task对象里，或者在冒泡到事件循环后，被默认的处理器（通常是打印到sys.stderr）处理了，但你可能没有注意到。
这种行为模式，是异步非阻塞I/O的副作用。为了最大化并发，事件循环不会因为一个任务的失败而停下所有其他任务。它需要一种机制来隔离这些失败，并在适当的时候让你能检查它们。
编程方式捕获和处理协程异常，核心思路就是把异步操作看作是返回一个“未来结果”的承诺。这个承诺可能是成功，也可能是失败（即异常）。
最直接的await与try...except：
当你的代码直接await一个协程时，这个协程抛出的任何未捕获的异常都会像同步函数一样传播到await它的位置。这是最简单、最符合直觉的处理方式。
import asyncio
async def might_fail():
print("Might fail coroutine running...")
await asyncio.sleep(0.1)
if True: # Simulate a condition that causes failure
raise ConnectionError("Failed to connect!")
return "Success"
async def caller_coroutine():
try:
result = await might_fail()
print(f"Coroutine succeeded with: {result}")
except ConnectionError as e:
print(f"Caught specific error: {e}")
except Exception as e:
print(f"Caught general error: {e}")
finally:
print("Cleanup after coroutine call.")
# asyncio.run(caller_coroutine())这种方式非常推荐，因为它让异常处理路径清晰明了。
使用asyncio.gather()和return_exceptions=True：
当你需要同时运行多个协程，并且希望即使其中一个协程失败，也不影响其他协程的执行，同时还能收集所有协程的结果（包括异常），asyncio.gather()配合return_exceptions=True就非常有用。
import asyncio
async def fetch_data(url):
print(f"Fetching data from {url}...")
await asyncio.sleep(0.2)
if "bad" in url:
raise ValueError(f"Invalid URL: {url}")
return f"Data from {url}"
async def run_multiple_fetches():
urls = ["http://good.com/data1", "http://bad.com/data2", "http://good.com/data3"]
tasks = [fetch_data(url) for url in urls]
# return_exceptions=True 确保即使有任务失败，gather 也不会中断，而是将异常作为结果返回
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, res in enumerate(results):
if isinstance(res, Exception):
print(f"Task {i} for {urls[i]} failed with: {type(res).__name__}: {res}")
else:
print(f"Task {i} for {urls[i]} succeeded with: {res}")
# asyncio.run(run_multiple_fetches())这里，gather会将异常本身作为结果列表中的一个元素返回，而不是直接抛出。这给了你灵活处理每个任务结果的机会。
Task.add_done_callback()：
前面在解决方案里已经提到了，这是一种非常强大的机制。当你启动一个后台任务，不打算直接await它，但又想知道它最终的状态时，回调函数是理想选择。回调函数会在任务完成时被调用，无论任务是成功、失败还是被取消。
import asyncio
async def background_job(job_id):
print(f"Background job {job_id} started...")
await asyncio.sleep(0.3)
if job_id % 2 != 0: # Simulate odd jobs failing
raise Exception(f"Job {job_id} failed!")
return f"Job {job_id} completed successfully."
def job_completion_handler(task):
job_id = task.get_name().split('_')[-1] # Assuming name like "Job_1"
if task.exception():
print(f"Handler: Job {job_id} finished with an error: {task.exception()}")
else:
print(f"Handler: Job {job_id} finished successfully with result: {task.result()}")
async def main_with_background_jobs():
for i in range(1, 5):
task = asyncio.create_task(background_job(i), name=f"Job_{i}")
task.add_done_callback(job_completion_handler)
print("Main is running, background jobs are being scheduled...")
await asyncio.sleep(1) # Give background jobs time to run and trigger callbacks
# asyncio.run(main_with_background_jobs())这种模式将任务的执行和结果处理解耦，非常适合构建复杂的异步系统。
在我看来，绝对是最佳实践。这不仅仅是一个好的做法，在构建健壮、可靠的asyncio应用时，它几乎是不可或缺的。
为什么这么说呢？
首先，尽管我们努力在每个协程内部使用try...except，或者在await的地方捕获异常，但总会有“漏网之鱼”。比如，一个回调函数本身在事件循环中被调用时抛出了异常，或者一个任务被创建后，你没有机会去await它，也没有给它添加done_callback。这些异常最终都会冒泡到事件循环的顶层。如果没有一个全局处理器，这些异常可能仅仅是打印到sys.stderr（asyncio的默认行为），然后事件循环继续运行，你的程序可能在不知情的情况下带着一个潜在的错误状态继续工作，这在生产环境中是灾难性的。
其次，全局异常处理器提供了一个集中式的异常监控点。你可以：
它就像你应用程序的最后一道防线。它不会替代局部的try...except（局部处理是更细粒度的控制），但它能捕获所有那些滑过指缝的错误，防止它们导致更隐蔽、更难以诊断的问题。
如何配置？很简单：
import asyncio
import logging
import sys
# 配置日志
logging.basicConfig(level=logging.ERROR, stream=sys.stderr,
format='%(asctime)s - %(levelname)s - %(message)s')
def global_exception_handler(loop, context):
"""
自定义的全局事件循环异常处理器
"""
exception = context.get("exception")
message = context.get("message", "An unhandled exception occurred in the event loop.")
task = context.get("task")
if exception:
logging.error(f"Global exception handler caught: {message}", exc_info=exception)
else:
logging.error(f"Global exception handler caught: {message}")
if task:
logging.error(f"Related task: {task.get_name() if task else 'Unnamed Task'}")
# 根据需要，你可以在这里执行更多操作，例如：
# if isinstance(exception, CriticalError):
# loop.stop() # 停止事件循环
# sys.exit(1) # 退出程序
async def some_risky_operation():
print("Executing some risky operation...")
await asyncio.sleep(0.1)
raise ValueError("Oops, this went wrong and wasn't caught locally!")
async def main_application():
loop = asyncio.get_running_loop()
loop.set_exception_handler(global_exception_handler)
# 故意创建一个不被await的任务，让其异常冒泡
asyncio.create_task(some_risky_operation(), name="RiskyTask")
print("Main application running, waiting for errors...")
await asyncio.sleep(0.5) # 给任务时间去执行和抛出异常
print("Main application finished.")
# asyncio.run(main_application())通过loop.set_exception_handler(your_handler_function)，你就可以把这个安全网部署到你的异步应用中了。它提供的context字典包含了异常、相关的任务（如果适用）以及其他上下文信息，让你能够进行细致的错误分析。这真的是异步服务稳定运行的基石之一。
以上就是Python怎样发现未处理的协程异常？的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
459
收藏
