当前位置：首页 > 文章列表 > 文章 > python教程 > Pythonasyncio异常处理技巧详解

Pythonasyncio异常处理技巧详解

2026-01-14 11:32:48 0浏览收藏

在文章实战开发的过程中，我们经常会遇到一些这样那样的问题，然后要卡好半天，等问题解决了才发现原来一些细节知识点还是没有掌握好。今天golang学习网就整理分享《Python asyncio 异常处理全攻略》，聊聊，希望可以帮助到正在努力赚钱的你。

使用 try-except 捕获 await 异常，create_task 需显式 await 或检查异常，gather 默认中断任务但可配置，wait 需手动检查，全局处理器用于监控未捕获异常。

Python asyncio 中的异常捕获技巧

在使用 Python 的 asyncio 编程时，异常处理比同步代码更复杂。协程可能在事件循环中异步执行，异常不会自动冒泡到主线程，若不妥善处理，会导致程序静默失败或资源泄漏。掌握正确的异常捕获方式，是编写健壮异步应用的关键。

直接 await 协程时的异常捕获

当你直接调用并 await 一个协程对象时，可以像处理普通函数一样使用 try-except 捕获异常。

协程内部抛出的异常会在 await 表达式处重新抛出
使用 try-except 包裹 await 语句即可捕获

示例：

import asyncio
<p>async def risky_task():
await asyncio.sleep(1)
raise ValueError("出错了")</p><p>async def main():
try:
await risky_task()
except ValueError as e:
print(f"捕获异常: {e}")</p><p>asyncio.run(main())
</p>

通过 create_task 启动任务的异常处理

使用 asyncio.create_task() 创建的任务会在后台运行，即使协程抛出异常，也不会立即被捕获。必须显式等待任务或检查其状态。

未被 await 的任务异常会滞留在任务对象中
任务异常只有在调用 result() 或 await 时才会触发

正确做法：保留任务引用并在合适时机 await 或检查异常。

async def main():
    task = asyncio.create_task(risky_task())
    try:
        await task
    except ValueError as e:
        print(f"任务异常: {e}")

如果不 await 任务，异常可能直到程序结束才被打印（通过 loop.set_exception_handler 可定制行为）。

并发执行中的异常捕获（gather 与 wait）

当需要同时运行多个协程时，常用 asyncio.gather 和 asyncio.wait，它们对异常的处理方式不同。

asyncio.gather 默认遇到第一个异常就停止其他任务（可配置 return_exceptions=True 改变行为）：

async def main():
    try:
        await asyncio.gather(
            good_task(),
            risky_task()
        )
    except ValueError as e:
        print(f"Gather 捕获: {e}")  # 立即中断

开启 return_exceptions=True 后，异常作为结果返回，不会中断其他任务：

results = await asyncio.gather(
    good_task(),
    risky_task(),
    return_exceptions=True
)
for r in results:
    if isinstance(r, Exception):
        print(f"任务异常: {r}")

asyncio.wait 返回完成和未完成的任务集合，需手动检查每个完成任务的异常：

done, pending = await asyncio.wait([task1, task2], return_when=asyncio.FIRST_EXCEPTION)
for t in done:
    if t.exception():
        print(f"任务出错: {t.exception()}")

全局异常处理器

可以通过设置事件循环的异常处理器来捕获未被处理的任务异常。

def custom_exception_handler(loop, context):
    msg = context.get("exception", context["message"])
    print(f"全局捕获异常: {msg}")
<p>loop = asyncio.get_event_loop()
loop.set_exception_handler(custom_exception_handler)
</p>

这适用于监控后台任务的意外崩溃，但不能替代正常的异常处理逻辑。

基本上就这些。关键是要意识到 async/await 中的异常不会自动传播，必须主动 await 任务或检查其结果。合理使用 try-except、gather 的参数以及全局处理器，才能构建可靠的异步系统。

今天关于《Pythonasyncio异常处理技巧详解》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！