await task 是捕获 asyncio 任务异常的硬性前提,因 Task 将异常锁在内部,只有 await(等价于 task.result())才会重新抛出;不 await 则异常静默丢失,不检查 task.exception() 则错误埋入日志死角。

不 await 任务,异常就静默丢失;不检查 task.exception(),错误就埋进日志死角。捕获 asyncio 并发异常的核心不是“加 try”,而是控制异常何时、以何种形式暴露出来。
await task 是捕获异常的硬性前提
asyncio 的 Task 对象把异常锁在内部,直到你显式触发它。await task 等价于调用 task.result(),而后者会重新抛出原始异常。常见错误是只创建任务、不 await:
task = asyncio.create_task(some_coro()) await asyncio.sleep(1) # 异常此时已发生,但不会冒泡
这种写法下,即使 some_coro() 抛了 ValueError,也不会进入任何 except 块。
- 必须对每个要捕获异常的任务做
await task - 若并发多个任务,用
asyncio.gather(*tasks)更安全(默认return_exceptions=False,任一失败即中断) - 若想“失败不中断”,则显式传
return_exceptions=True,之后手动检查结果中是否为异常实例
asyncio.gather(return_exceptions=True) 怎么用才不踩坑
启用 return_exceptions=True 后,gather 不再中断流程,而是把异常作为返回值之一塞进结果列表。但容易忽略的是:它不区分异常类型,也不自动展开嵌套结构。
- 结果中每个元素可能是正常返回值,也可能是异常对象(如
ValueError、TimeoutError),必须用isinstance(r, Exception)判断 - 若协程里又
await了另一个可能失败的协程,且没做内层处理,异常仍会透出到gather结果中 - 不要在
return_exceptions=True模式下还包一层try-except去捕gather调用本身——它不会抛异常,只会返回异常对象
后台任务(如心跳、上报)怎么查异常而不阻塞
有些任务你不能或不该立刻 await(比如长期运行的后台心跳),这时得靠 task.exception() 主动捞异常——它不抛错,只返回异常对象或 None。
- 先判断
task.done(),再调task.exception();否则返回None,不代表没异常,只是还没结束 -
task.exception()返回值可能是:None(正常完成)、异常实例(失败)、或尚未结束(此时done()为False) - 不要用
task.result()替代——它会重抛异常,可能崩掉当前协程 - 推荐搭配
add_done_callback使用,避免轮询:task.add_done_callback(lambda t: print(t.exception()))
TaskGroup 中多个异常怎么分类型处理
asyncio.TaskGroup 允许同时启动多个任务,并在退出时聚合所有异常。但它不按“第一个抛错就停”的方式工作,而是等全部任务结束(或被取消)后,把异常打包成 ExceptionGroup 或子类抛出。
- 用
except* ValueError捕获所有ValueError实例(哪怕有多个) - 用
except* Exception捕获其余所有异常,其.exceptions属性是元组,每个元素是原始异常 - 避免只写
except Exception,它会漏掉ExceptionGroup——因为外层不是单个ValueError,而是一个容器 - 若需记录每个异常的原始 traceback,得遍历
e.exceptions并分别打印
最易被忽略的一点:异常是否被“retrieve”(取回),决定了它会不会触发 loop.set_exception_handler。未 await、未调 exception()、未注册 done callback 的任务,一旦出错,就真的消失了——连日志都留不下痕迹。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











