asyncio.run() 无法直接捕获 create_task 启动的任务异常,需在任务内处理或通过 await、gather(return_exceptions=true)、task.exception() 显式获取;retrying 不支持异步。

asyncio.run() 里直接用 try/except 捕获不到任务异常
异步任务抛出的异常不会自动冒泡到 asyncio.run() 外层,尤其当任务是通过 asyncio.create_task() 启动后——它会静默失败,控制台可能只打印一句 Task exception was never retrieved,然后就没了。
真正要捕获异常,得在任务内部处理,或显式等待任务完成:
- 用
await task(适合串行场景),此时异常会正常抛出 - 用
asyncio.gather(..., return_exceptions=True)收集结果和异常,再统一判断 - 对单个
Task对象调用task.exception()主动检查(需确保已 done)
retrying 库不支持 async/await,别往协程里硬塞
像 retrying 或老版本 tenacity(RuntimeWarning: coroutine 'xxx' was never awaited。
必须用原生支持异步的重试方案:
立即学习“Python免费学习笔记(深入)”;
-
tenacity≥ 8.0:直接用@retry装饰协程,内部自动await - 手写重试逻辑:用
while True:+try/except+await asyncio.sleep(),注意加退出条件(如最大重试次数) - 避免用
time.sleep(),它会阻塞整个协程调度
重试时忘记 await 依赖协程,导致“假成功”
常见错误是重试逻辑写了,但里面调用的下游协程没加 await,比如:
async def fetch_data():
return requests.get(...) # ❌ requests 是同步库,这里根本没 await,还可能报错或者更隐蔽的:
async def do_work():
result = some_async_func() # ❌ 忘了 await,result 变成 coroutine 对象,类型检查都过不去
if not result:
raise ValueError("empty")这类问题不会立刻报错,但重试判断基于错误值(比如 None 或空字符串),实际根本没发请求。务必确认每个异步调用前都有 await。
超时和重试组合使用时,timeout 要包住整个重试块
如果只给单次请求设 asyncio.wait_for(..., timeout=5),重试 3 次就可能耗时 15 秒;但业务要求“最多等 6 秒”,就得把重试逻辑整体包进一个外层 timeout:
try:
await asyncio.wait_for(perform_with_retry(), timeout=6)
except asyncio.TimeoutError:
raise TimeoutError("total retry time exceeded 6s")否则容易出现:单次请求 2 秒 + 等待间隔 1 秒 × 3 次 = 实际耗时远超预期,下游服务等不及就断连了。
重试间隔建议用指数退避(如 2 ** attempt * 0.1),别固定 1 秒——并发高时容易集体重试,压垮下游。










