功能说明
retry 是一个参数化装饰器,用于包装可能瞬时失败的操作。构造时可指定最大尝试次数、基础延迟、延迟上限、抖动比例、可重试异常元组与休眠函数。参数在装饰阶段就完成校验,非法值立即抛出 RetryConfigError。被包装的函数每次失败后,若还有剩余尝试,就按 min(cap_s, base_delay_s * 2 ** attempt) 计算延迟,再加一个 uniform(0, delay * jitter_ratio) 的随机量,最后调用注入的 sleep。用尽尝试后重新抛出最后一次异常。
关键代码
退避与抖动的计算只有三行:
delay = min(cap_s, base_delay_s * (2 ** attempt))
jitter = random.uniform(0, delay * jitter_ratio)
sleep(delay + jitter)
min 保证延迟不会无限增长;抖动只做正向叠加,因此实际等待落在 delay 与 delay * (1 + jitter_ratio) 之间。sleep 默认是 time.sleep,测试时可替换为记录函数,从而无需真实等待。@wraps(fn) 保留原函数的名称与文档字符串。
使用注意
- 默认
retryable=(Exception,)会把编程错误也纳入重试,实际使用应传入具体异常类型。 - 仅支持同步函数,协程需要另写异步版本或改用支持 async 的库。
- 没有总时长上限,最坏情况下累计等待可能远超预期。
- 抖动是加在延迟之后的正向偏移,属于 decorrelated 之外的简单策略,不是完整抖动。
- 结尾的
assert last_error is not None在python -O下会被移除,此时可能抛出None导致TypeError。
验证方式
源码测试覆盖四项:瞬时失败三次后成功、非可重试异常不重试、延迟序列被上限截断为 [1.0, 2.0, 2.0]、非法参数被拒绝。测试通过传入 sleep=lambda s: None 或 sleep=delays.append 避免真实等待。保存实现后运行 python -m unittest 即可;也可用一个计数函数模拟前两次抛 TimeoutError、第三次返回成功来观察调用次数。