Python asyncio 中的异常捕获技巧_石家庄创泽智能科技有限公司

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Python asyncio 中的异常捕获技巧

2025-11-20

浏览次数：次

返回列表

使用 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())

通过 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，它们对异常的处理方式不同。

Python v2.4 中文手册 chm

Python v2.4版chm格式的中文手册，内容丰富全面，不但是一本手册，你完全可以把她作为一本Python的入门教程，教你如何使用Python解释器、流程控制、数据结构、模板、输入和输出、错误和异常、类和标准库详解等方面的知识技巧。同时后附的手册可以方便你的查询。

2 查看详情 Python v2.4 中文手册 chm

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)