可以实现支持自动重试的async装饰器，核心是捕获异常、控制重试次数与间隔，并在异步上下文中正确await目标函数；基础版固定次数立即重试，增强版支持指数退避、抖动及指定异常类型。
可以实现一个支持自动重试的 async 装饰器，核心是捕获异常、控制重试次数与间隔，并在异步上下文中正确 await 目标函数。
基础版本：固定重试次数 + 立即重试
最简实现，适用于快速验证逻辑或对延迟不敏感的场景：
import asyncio
from functools import wraps
<p>def retry(max_attempts=3):
def decorator(func):
@wraps(func)
async def wrapper(*args, *<em>kwargs):
for attempt in range(max_attempts):
try:
return await func(</em>args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise e
return None # 不会执行到这里，仅作类型提示补充
return wrapper
return decorator</p><h1>使用示例</h1><p>@retry(max_attempts=3)
async def fetch_data():
await asyncio.sleep(0.1)
raise ConnectionError("Network failed")
增强版本：指数退避 + 可配置异常类型
生产环境更常用——避免重试风暴，且只对特定异常重试（如网络错误），跳过逻辑错误（如 ValueError）：
import asyncio
import random
from functools import wraps
<p>def retry(
max_attempts=3,
delay=1.0,
backoff=2.0,
jitter=True,
exceptions=(Exception,)
):
def decorator(func):
@wraps(func)
async def wrapper(*args, *<em>kwargs):
current_delay = delay
for attempt in range(max_attempts):
try:
return await func(</em>args, **kwargs)
except exceptions as e:
if attempt == max_attempts - 1:
raise e</p><h1>指数退避 + 可选抖动</h1><pre class="brush:php;toolbar:false;"> if jitter:
current_delay *= (0.5 + random.random() / 2)
await asyncio.sleep(current_delay)
current_delay *= backoff
return None
return wrapper
return decorator只对网络类异常重试
@retry( max_attempts=5, delay=0.5, exceptions=(ConnectionError, asyncio.TimeoutError, OSError) ) async def api_call(): ...
实用技巧与注意事项
- 避免重试幂等性问题：不要对有副作用的非幂等操作（如支付扣款、发邮件）盲目重试，应先确认接口是否支持重试语义
-
区分异常类型很关键：用元组指定
exceptions，例如(TimeoutError, aiohttp.ClientError)，避免吞掉KeyboardInterrupt或SystemExit - 日志记录建议加在装饰器内：每次重试前打印日志，便于排查；可传入 logger 实例增强灵活性
-
超时控制需配合使用：单独重试不够，建议在被装饰函数内部或调用处加上
asyncio.wait_for(..., timeout=...)
