Python异步任务默认不会自动重试，一旦抛出未捕获异常便静默失败，且异常不向上冒泡，导致主程序完全无法感知子任务错误；本文深入剖析了手动实现可靠重试的必要性与核心陷阱，详解如何用自定义 `@async_retry` 装饰器安全支持异常过滤、指数退避和超时兜底，并强调必须使用 `await asyncio.sleep()` 替代阻塞式 `time.sleep()`、避免同步库阻塞事件循环、明确区分单次请求超时与整体重试时限——真正考验功力的不是代码实现，而是对业务语义的精准把握：什么错误值得重试？何时该果断放弃？

不会自动重试。asyncio 本身不提供任何内置重试机制，asyncio.create_task() 启动的任务一旦抛出未捕获异常，只会静默失败，控制台可能只打印 Task exception was never retrieved，然后就没了。

asyncio 任务异常默认不传播，重试必须手动编码

这是最常被忽略的前提：异步任务的异常不会像同步函数那样冒泡到调用栈顶层。你写 asyncio.run(main())，但 main() 里用 asyncio.create_task(fetch()) 启动的协程，就算 fetch() 抛了 ConnectionError，asyncio.run() 也完全感知不到。

• 必须在任务内部用 try/except 捕获，或显式调用 task.exception() 检查
• asyncio.gather(..., return_exceptions=True) 可让失败任务返回异常对象而非中断整个集合
• 别依赖 asyncio.run() 的返回值判断成败——它只反映 main() 协程是否成功，不管子任务

@retry 装饰器不能直接套 async 函数

常见错误是把同步重试装饰器（比如用 time.sleep() + for 循环）直接套在 async def 上。结果就是：协程对象被当成普通函数执行，await 缺失，返回的是 coroutine object 而非实际结果，后续逻辑全错乱。

• 装饰器内必须用 await 调用目标协程，不能用 time.sleep()，得用 await asyncio.sleep()
• 重试间隔建议用指数退避，例如 await asyncio.sleep(2 ** attempt * 0.1)，避免并发请求集体重试压垮下游
• 若需支持超时约束（如“整个重试过程最多耗时 6 秒”），得在外层包一层 asyncio.wait_for()，而不是只给单次请求设 timeout

手写 async retry 装饰器的关键点

一个可用的最小可行版本要处理三件事：异常过滤、等待策略、超时兜底。下面这个例子能跑通，且避开大部分坑：

import asyncio
import functools
def async_retry(max_attempts=3, exceptions=(Exception,), backoff_factor=0.1):
def decorator(func):
@functools.wraps(func)
async def wrapper(*args, kwargs):
last_exc = None
for attempt in range(max_attempts):
try:
return await func(*args, *kwargs)
except exceptions as e:
last_exc = e
if attempt < max_attempts - 1:
delay = backoff_factor  (2  attempt)
await asyncio.sleep(delay)
raise last_exc
return wrapper
return decorator
用法示例
@async_retry(max_attempts=3, exceptions=(ConnectionError,), backoff_factor=0.2)
async def fetch_data(url):
这里应使用 aiohttp 或 httpx，不是 requests
raise ConnectionError("simulated network failure")

• 装饰器函数名带 async_ 前缀，明确区分同步/异步版本
• exceptions 参数必须是元组，否则 isinstance(e, exceptions) 会报错
• 别在装饰器里混用 requests 等同步库——它会阻塞整个事件循环
• 如果业务要求“总耗时不超 N 秒”，得在调用处加 await asyncio.wait_for(fetch_data(...), timeout=N)，装饰器本身不负责这个

真正难的不是写出来，而是想清楚重试边界：哪些异常该重试（ConnectionError 可以，ValueError 通常不该），重试间隔怎么随次数增长，以及超时是按单次还是整体计算——这些没对齐业务语义，重试反而会放大问题。

本篇关于《Python异步任务能重试吗？自定义装饰器实现》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！