给重试逻辑加一层护栏:写一个可复用的 Python 重试装饰器
【摘要】 重试不是无脑循环:用一个可复用的 Python 装饰器,把重试次数、指数退避和抖动集中管理,只重试瞬时错误,给线上调用加一层护栏。
调用外部接口、写数据库或者发消息时,偶尔遇到网络抖动、超时是很常见的事。直接让请求失败,用户体验不好;无脑地一直重试,又可能把一次小故障放大成雪崩。一个可复用的重试装饰器,就是给重试逻辑加一层护栏:把“什么时候重试、重试几次、每次等多久”集中到一处,而不是散落在每个业务函数里。
先定四条原则
- 只重试瞬时错误。网络超时、连接中断可以重试;参数错误、鉴权失败、数据校验失败不应该重试,重试多少次结果都一样。
- 一定要有上限。最大重试次数和单次等待上限都要写死,避免无限循环拖垮调用方。
- 用指数退避加抖动。第一次等 0.5 秒,第二次等 1 秒,第三次等 2 秒,同时加一点随机抖动,避免大量客户端在同一时刻一起重试。
- 保持幂等意识。重试的前提是这次操作重复执行是安全的,例如查询、幂等写入;非幂等的扣款、发券类操作不要直接套用通用重试。
一个可复用的实现
import functools
import random
import time
def retry(max_attempts=3, base_delay=0.5, max_delay=4.0,
retry_on=(ConnectionError, TimeoutError)):
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 retry_on as exc:
if attempt == max_attempts:
raise
sleep = delay + random.uniform(0, delay * 0.5)
print(f"第 {attempt} 次失败:{exc},{sleep:.2f} 秒后重试")
time.sleep(sleep)
delay = min(delay * 2, max_delay)
return wrapper
return decorator
怎么用
@retry(max_attempts=3, base_delay=0.5)
def fetch_data():
# 这里放真实的请求逻辑,例如调用某个 HTTP 接口
raise ConnectionError("网络抖动")
调用方不需要关心重试细节,只需要知道:这个函数在瞬时故障时会自己重试最多 3 次,超过次数后把原始异常抛出来,由上层统一处理。
上线前再检查三件事
- 重试次数和等待时间是否和接口的超时时间匹配,不要出现“重试还没结束,调用方已经超时”的尴尬。
- 是否记录了每次失败的原因和次数,否则线上排查时只能看到最后一次异常。
- 是否区分了可重试和不可重试的异常类型,清单要写在代码里,而不是只靠注释提醒。
把重试写成装饰器,好处不是代码变短,而是策略变统一:以后要改退避策略、加监控埋点,只改一处,所有被装饰的函数都会跟着变。这也是很多线上故障复盘里最容易被忽略的一环:重试本身没有错,错的是没有护栏的重试。
【声明】本内容来自华为云开发者社区博主,不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息,否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱:
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论(0)