为什么Python异步协程抛出的异常没有打印？如何全局捕获？-Python教程-PHP中文网

asyncio.run()中未await的后台任务异常会被静默丢弃，仅发警告；应通过set_exception_handler全局捕获、gather(return_exceptions=true)显式处理或启用debug=true调试。

为什么python异步协程抛出的异常没有打印？如何全局捕获？

asyncio.run() 里未处理的协程异常会被吞掉

Python 3.7+ 的 asyncio.run() 在主协程（即传入的第一个协程）抛出未捕获异常时，会正常打印 traceback 并退出；但如果你在它内部用 asyncio.create_task() 或 asyncio.ensure_future() 启动了其他“火种任务”（fire-and-forget tasks），这些任务里的异常默认不会传播，也不会打印——asyncio 会静默丢弃它们，只发一条警告：Task exception was never retrieved。

常见诱因包括：

忘记 await 一个 task，比如写了 asyncio.create_task(fetch_data()) 却没保存或 await 它
用 asyncio.gather(..., return_exceptions=False) 但没对结果做异常检查
在后台 task 里调用了可能失败的 IO 操作（如 HTTP 请求、DB 查询），又没包 try/except

给 asyncio 设置异常处理器：loop.set_exception_handler()

这是最直接的全局捕获方式，适用于所有未被 await 或未被 task.exception() 显式取走的 task 异常。注意：它只对 asyncio.Task 有效，不拦截普通协程（coroutine object）抛出的异常（那些得靠外层 await 捕获）。

实操建议：

必须在 event loop 创建后、运行前设置，通常就在 asyncio.run() 之前手动获取 loop 并配置
handler 函数接收两个参数：loop 和 context，其中 context['exception'] 是实际异常对象
别只 print，建议至少记录日志级别为 ERROR，并考虑是否要 os._exit(1) 防止程序带病运行

import asyncio
import logging
<p>def custom_exc_handler(loop, context):
exc = context.get('exception')
if exc:
logging.error('Uncaught exception in task', exc_info=exc)
else:
logging.error('Asyncio exception context without exception: %r', context)</p><p>loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_exception_handler(custom_exc_handler)</p><h1>然后才 run</h1><p>loop.run_until_complete(main())
</p>

避免依赖全局 handler：显式 await + gather(return_exceptions=True)

真正健壮的做法不是靠兜底，而是让异常“浮上来”。对一组并发任务，优先用 asyncio.gather() 而非一堆 create_task()，并设 return_exceptions=True。这样即使某个子协程失败，gather 也不会中断，而是把异常作为对应位置的返回值，你可以在后续统一检查。

python全能编程助手

SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的双功能合体技能，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途：；核心要点📝 题解模式：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式：输入 Python 代码，自动添加详细中。它将相关步骤、

下载

对比示例：

# ❌ 静默失败风险高
tasks = [asyncio.create_task(coro()) for coro in coros]
# 忘记 await tasks → 异常丢失
<h1>✅ 可控、可检查</h1><p>results = await asyncio.gather(*coros, return_exceptions=True)
for i, r in enumerate(results):
if isinstance(r, Exception):
logging.error('Task %d failed: %r', i, r)
</p>

注意：return_exceptions=False（默认）下，第一个异常就会让整个 gather 抛出，后续任务可能被取消——这本身也是种控制流，但需要你明确设计错误传播策略。

调试阶段加临时钩子：监听 asyncio.Task 的 _log_traceback

开发时想快速看到谁漏了异常？可以 monkey patch asyncio.Task.__init__ 或更稳妥地，在测试环境启用 asyncio 的调试模式：asyncio.run(main(), debug=True)。这时，一旦有 task 异常未被 retrieve，会强制打印完整 traceback，并附上 task 创建时的栈帧（含文件名和行号）。

但注意两点：