Python 重试装饰器实战:指数退避、异常分类与可观测日志
【摘要】 区分瞬时失败与确定性失败,用指数退避加随机抖动控制重试节奏,配合次数上限与结构化日志,给出可直接复用的 Python 重试装饰器。
问题:为什么不能无脑重试
调用外部接口时,网络抖动、限流和服务端短暂过载是常见瞬时失败;但参数错误、鉴权失败、数据校验失败重试多少次也不会成功。把两类失败混在一起重试,既浪费时间,又可能把限流打成雪崩。正确做法是先分类,再决定是否重试、重试几次、间隔多久。
核心设计
- 只对瞬时异常重试,例如超时、连接重置、HTTP 429 和 503;对 400、401、403、404、422 直接抛出。
- 间隔用指数退避:第一次 0.5 秒,第二次 1 秒,第三次 2 秒,并加随机抖动,避免大量客户端在同一时刻齐刷刷重试。
- 设置总次数上限和总耗时上限,超过就放弃并记录最后一次异常,便于排查而不是无限挂起。
- 每次重试都写一条结构化日志:函数名、已重试次数、下一次间隔、异常类型。线上排障时只看这几行就能还原全过程。
可直接复用的实现
import functools, random, time
RETRYABLE = (TimeoutError, ConnectionError)
def retry(max_attempts=4, base_delay=0.5, max_delay=8.0):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
delay = base_delay
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except RETRYABLE as exc:
if attempt == max_attempts:
raise
sleep = min(max_delay, delay) + random.uniform(0, 0.3)
print(f"{func.__name__} 第 {attempt} 次失败:{exc!r},{sleep:.2f} 秒后重试")
time.sleep(sleep)
delay *= 2
return None
return wrapper
return decorator
注意:上面对异常类型的判断要按真实业务扩展,例如 requests 的异常和业务方约定的错误码。把“可重试”做成参数传入装饰器,比在装饰器里写死一串异常更易复用。
使用示例与检查清单
@retry(max_attempts=4)
def fetch_user(user_id):
# 这里调用真实接口;超时或连接异常会触发重试
raise TimeoutError("upstream timeout")
上线前逐项核对:接口是否幂等(写操作重试前必须确认幂等键或服务端去重);重试是否会放大写副作用;限流场景下退避是否足够长;日志里是否包含请求标识,便于和服务端日志对齐。
结论
重试不是“多试几次就好”,而是一组有边界的策略:分类决定能不能重试,退避决定何时重试,上限决定何时放弃,日志决定事后能不能讲清楚。把这四点写进一个小装饰器,多数接口的健壮性就能上一个台阶。
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)