asyncio.run()中顶层协程未捕获异常会强制退出，无法用外层try/except捕获；需用create_task()+add_done_callback、gather(return_exceptions=True)或TaskGroup（3.11+）实现可控异常处理。
asyncio.run() 里异常直接崩溃，根本捕不到
默认用 asyncio.run() 启动协程时，只要顶层协程抛出未处理异常（比如 ValueError 或 ConnectionError），整个事件循环就终止，且异常不会被外层 try/except 捕获。这不是“漏捕”，而是 asyncio 的设计行为：它把未处理异常当作程序级错误强制退出。
常见现象：RuntimeWarning: coroutine 'xxx' was never awaited 或直接进程退出，except Exception 完全无效。
- 必须用
asyncio.create_task()显式提交协程到事件循环，才能对单个任务做异常监听 -
asyncio.run()只适合最简单的脚本入口，不适用于需容错的协程编排 - 若坚持用
asyncio.run()，唯一办法是确保顶层协程自己try/except—— 但这就不是“跨协程传播”的场景了
用 asyncio.create_task() + done_callback 拦截子协程异常
子协程（如通过 asyncio.create_task() 启动的）如果抛出未处理异常，不会立刻冒泡，而是被封装进 Task 对象内部，直到你主动检查或等待它完成。这时可以用 add_done_callback() 注册回调，在任务结束时读取异常。
import asyncio
<p>async def risky():
await asyncio.sleep(0.1)
raise ValueError("boom")</p><p>def handle_task_result(task):
if task.exception():
print("捕获到异常:", task.exception())</p><h1>这里可记录、重试、通知等</h1><pre class="brush:php;toolbar:false;">else:
print("结果:", task.result())async def main(): task = asyncio.create_task(risky()) task.add_done_callback(handle_task_result) await task # 等待完成，触发 callback
asyncio.run(main())
-
task.exception()返回None表示无异常，否则返回实际异常对象 - 回调函数必须接受一个
Task参数，不能带额外参数；需用闭包或类属性传上下文 - 注意：回调在事件循环线程中同步执行，不要在里面做阻塞操作（如
time.sleep、文件写入）
asyncio.gather() 的 return_exceptions=True 是最简方案
当你并发启动多个协程并用 asyncio.gather() 等待时，只要有一个失败，默认整组失败并抛出第一个异常。加 return_exceptions=True 后，失败协程的结果变成异常对象，成功协程的结果保持原值，全部统一返回为列表。
async def a(): return "ok1"
async def b(): raise KeyError("missing")
async def c(): return "ok2"
<p>results = await asyncio.gather(a(), b(), c(), return_exceptions=True)</p><div class="aritcle_card flexRow artxards">
<div class="artcardd flexRow">
<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill5772" title="Python Use Agent"><img
src="https://img.php.cn/upload/skill/000/000/081/179065807481489.jpg" alt="Python Use Agent" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
<div class="aritcle_card_info flexColumn">
<a rel="nofollow" href="/xiazai/skill5772" title="Python Use Agent" class="overflowclass">Python Use Agent</a>
<p class="overflowclass">智能执行Python任务，自动生成、执行代码并反馈结果，无需额外配置，兼容旧命令。</p>
</div>
<a rel="nofollow" href="/xiazai/skill5772" title="Python Use Agent" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
</a>
</div>
</div><h1>results == ["ok1", KeyError('missing'), "ok2"]</h1><p></p>
- 适合批量任务、不需要实时响应失败的场景（如并发请求多个 API）
- 返回的是异常实例，不是字符串，可直接
isinstance(r, Exception)判断 - 不适用于需要按任务粒度做不同恢复策略的情况（比如对网络异常重试、对校验异常跳过）
使用 asyncio.TaskGroup（Python 3.11+）做结构化异常管理
asyncio.TaskGroup 是目前最接近“跨协程异常传播可控”的方案：它允许你在一组协程中局部处理异常，而不影响其他仍在运行的任务，且能区分“哪个子任务失败”。
async def worker(name):
if name == "bad":
raise RuntimeError("crashed in worker bad")
await asyncio.sleep(0.1)
return f"done {name}"
<p>async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(worker("good"))
tg.create_task(worker("bad"))
tg.create_task(worker("also-good"))
except<em> RuntimeError as eg: # 注意：是 except</em>，不是 except
print("捕获到 RuntimeError 组:", eg.exceptions)</p><p>asyncio.run(main())
</p>
- 必须用
except*（异常匹配组）语法，普通except捕不到 - 失败任务会取消同组其他正在运行的任务（可配置
shield=True避免） - Python 3.11 才引入，旧版本不可用；3.11.2+ 修复了部分
TaskGroup在取消时的竞态问题
真正难处理的，是那些没被显式 await、也没被 create_task() 接管的“幽灵协程”——比如回调里忘了 await 一个协程，或者信号处理函数里直接调了异步函数。它们的异常会静默丢失，连 sys.unraisablehook 都不一定触发。这种只能靠静态检查（如 pylint 的 await-outside-async 规则）和单元测试覆盖路径来提前暴露。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
