写服务端代码几乎免不了和外部系统打交道,调别人的 HTTP 接口、读写数据库、发消息队列,偶尔都会因为网络抖动、对方临时限流、连接超时而失败。这类失败很多是暂时的,过上一小会儿再试一次就成功了,可如果代码一遇异常直接抛出去,一次小小的抖动就让整个任务失败,用户那边看到的就是一个莫名其妙的报错。重试就是对付这类瞬时故障的标准手段,它把偶发失败挡在用户看不见的地方。代价是你得想清楚什么时候该停、隔多久再试、只对哪些异常重试,瞎重试比不重试还危险。
Tenacity 是 Python 里用装饰器实现的重试库,给函数加上 @retry,就能把这些规则声明式地写在函数头上。这一节我们从它最朴素的写法讲起,逐个拆开三个旋钮,分别是停止条件、等待间隔和异常筛选,最后用一个调用外部订单接口的完整例子把三者组合起来,再说清楚哪些情况反而不该重试。读之前只需要知道 Python 的函数装饰器是什么,明白 try except 捕获异常的基本用法。
安装只要一行命令:
pip install tenacity
1. 朴素写法
先看 Tenacity 最简单的用法,不加任何参数,直接给函数挂一个 @tenacity.retry:
import tenacity
@tenacity.retry
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
if __name__ == '__main__':
never_gonna_give_you_up()
这段代码里 never_gonna_give_you_up 每次都主动 raise Exception,装饰器抓到异常后什么都不等,立刻把函数再跑一遍,于是它会一直打印下去,直到进程被你手动杀掉。这就是不加参数的默认行为,任何异常都触发重试,重试之间也不等待。它适合在命令行脚本里快速试错,放到生产环境却是个隐患,因为一旦失败是永久性的,它会把 CPU 空转到底。要让它可控,就得先回答一个问题:什么时候才算试够了。
2. 停止条件
默认的重试永不停止,生产代码必须给它一个放弃的理由。Tenacity 用 stop 参数控制这件事,你可以按次数停,也可以按时间停,两个还能组合。下面逐个看。
2.1 限制重试次数
最直觉的做法是限定最多试几次,用 stop_after_attempt:
import tenacity
@tenacity.retry(stop=tenacity.stop_after_attempt(5))
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
if __name__ == '__main__':
never_gonna_give_you_up()
stop_after_attempt(5) 的意思是,连最初那次算在内一共尝试 5 次,5 次之后仍然抛异常,Tenacity 就不再重试,转而抛出自己的 RetryError,把失败交还给调用方。这个 RetryError 长这样:
tenacity.RetryError: RetryError[<Future at 0x7fe1300c9050 state=finished raised Exception>]
为什么要按次数停?因为很多失败根本重试不好,比如参数填错了、权限不够,这类错误试 1 万次结果都一样。给次数设个上限,至少能保证程序不会陷在里面出不来。
2.2 限制总耗时
按次数停解决了无限循环,可有的场景你关心的不是试了几次,而是总共花了多久。比如一个接口要求在很短时间内返回,你不会在意它中间重试了几次,只在意从第一次调用算起总共拖了多久。stop_after_delay 就是按这个思路设计的:
import tenacity
@tenacity.retry(stop=tenacity.stop_after_delay(1))
def never_gonna_give_you_up():
print("Retry forever ignoring Exceptions, don't wait between retries")
raise Exception
if __name__ == '__main__':
never_gonna_give_you_up()
stop_after_delay(1) 表示从函数第一次被调用开始计时,超过 1 秒还没成功就放弃,单位是秒。它和按次数停是两种互补的思路,次数上限防的是失败永久不变,时间上限防的是整个请求被拖死。
2.3 两个条件并用
实际项目里这两个限制通常要一起上,既怕试太多次,又怕拖太久。Tenacity 允许用竖线把两个停止条件连起来,意思是哪个先触发就按哪个停:
import tenacity
@tenacity.retry(stop=(tenacity.stop_after_delay(5) | tenacity.stop_after_attempt(5)))
def stop_after_10_s_or_5_retries():
print("Stopping after 10 seconds or 5 retries")
raise Exception
if __name__ == '__main__':
stop_after_10_s_or_5_retries()
stop_after_delay(5) 竖线 stop_after_attempt(5) 表示 5 秒内还没成功,或者已经试了 5 次,哪条先到就先停。这里的竖线不是按位运算,是 Tenacity 重载过的逻辑或,两个条件满足任意一个即放弃。
3. 重试间隔
停的条件说清楚了,还有一个容易被忽略的问题:重试之间要不要等。默认情况下 Tenacity 抓到异常立刻重跑,中间不睡。这在对端只是偶发抖动时没问题,但如果对端已经因为过载在报错,你这边毫秒级地反复打过去,等于在它最忙的时候又压上去一把。更麻烦的是,成百上千个客户端如果都按同一个固定节奏重试,它们的重试会在同一秒撞在一起,把刚缓过来的服务再次打垮。所以重试之间通常要主动等一会儿,这件事由 wait 参数控制。
3.1 固定等待
最直接的是固定间隔,每次重试前都睡固定的秒数:
import tenacity
@tenacity.retry(wait=tenacity.wait_fixed(2))
def wait_2_s():
print("Wait 2 second between retries")
raise Exception
if __name__ == '__main__':
wait_2_s()
wait_fixed(2) 表示两次重试之间固定等 2 秒。它简单可预测,适合对端恢复时间比较稳定的场景。
3.2 随机等待
固定间隔的问题在于所有客户端整齐划一。wait_random 让每次等待落在一个随机区间里,把大家的重试时间打散:
import tenacity
@tenacity.retry(wait=tenacity.wait_random(min=1, max=2))
def wait_random_1_to_2_s():
print("Randomly wait 1 to 2 seconds between retries")
raise Exception
if __name__ == '__main__':
wait_random_1_to_2_s()
wait_random(min=1, max=2) 表示每次在 1 到 2 秒之间随机挑一个等待时长。同样是退避一秒多,每个客户端错开之后,打到服务端的时间就摊平了,不再齐刷刷地撞过去。这两个是最基础的等待策略,实战里还常用指数退避,也就是每次等待按倍数拉长,那个我们放到下一节的完整例子里一起看。
4. 异常筛选
前面几个例子还有个共同特点:只要函数抛出任何异常,Tenacity 都重试。这其实太宽了。你自己代码里的 KeyError、传错参数的 ValueError、接口返回 400 的请求错误,这些再试多少次都不会变,重试只会白白浪费时间,还把真正的 bug 掩盖在一次次重试里。真正值得重试的,是连接超时、读取失败这类外部因素引起的错误。Tenacity 用 retry 参数让你圈定重试范围:
import tenacity
@tenacity.retry(retry=tenacity.retry_if_exception_type(IOError))
def might_io_error():
print("Retry forever with no wait if an IOError occurs, raise any other errors")
raise IOError
if __name__ == '__main__':
might_io_error()
retry_if_exception_type(IOError) 的意思是,只在抛出 IOError 时才重试,换成别的异常,比如 ZeroDivisionError,就直接往外抛,Tenacity 不再接手。这样网络读写类的瞬时错误会被自动重试,而你自己逻辑里的错误会立刻暴露出来。实际项目里通常不止一种异常值得重试,可以传一个元组,把 ConnectionError、Timeout 等都列进去,下一节我们就这么用。
5. 实战例子
三个旋钮分别看明白了,现在放到一个真实场景里组合。假设你要调一个外部订单接口查订单详情,这个接口偶尔会连接超时或者连接被重置。你希望的行为是,只在这类网络错误上重试,最多试 3 次,每次等待按指数退避慢慢拉长,3 次都不行就放弃并把失败记录下来。完整代码如下,需要先 pip install requests:
import requests
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
RetryError,
)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, max=4),
retry=retry_if_exception_type((
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
)),
)
def fetch_order(order_id):
url = f"https://api.example.com/orders/{order_id}"
resp = requests.get(url, timeout=2)
resp.raise_for_status()
return resp.json()
if __name__ == "__main__":
try:
order = fetch_order(10086)
print("拿到订单", order)
except RetryError:
print("重试三次仍然失败,把这次请求记下来稍后人工处理")
我们逐行看这个配置。stop_after_attempt(3) 把总尝试次数压在 3 次,不会无限打下去。wait_exponential(multiplier=1, max=4) 让每次等待按指数倍拉长,大致逐次翻倍,同时用 max 封顶在 4 秒,不会越等越久失控。retry_if_exception_type 圈定了只在 ConnectionError 和 Timeout 上重试,代码里 resp.raise_for_status() 抛出的 HTTPError 不在名单里,所以接口返回 400 或 500 时它直接报错,不会被无谓地重试。最后函数跑在 try except 里,重试耗尽后 Tenacity 抛出 RetryError,这里打印一条日志,把这次失败交给后续的告警或人工流程。
6. 何时别重试
重试好用,但用错地方比不用更糟。有三类情况必须忍住别重。
第一类,永久性错误别重。接口返回 404 找不到资源、401 没权限、400 参数错误,这些是你这边请求本身有问题,再发一次结果不会变。用 retry_if_exception_type 把重试范围圈在网络类异常上,就是为了把这类错误挡在重试之外。
第二类,非幂等操作别瞎重。重试本质是把同一个请求再发一遍,如果这个操作会真的改数据,比如扣款、下单、发优惠券,第一次其实已经成功了只是响应超时,你再重试一次就可能重复扣款。这类操作要么带上幂等键让对端去重,要么干脆不自动重试。
第三类,自己代码的 bug 别靠重试掩盖。函数因为字典取键不存在、类型传错而崩,这是你的逻辑有问题,重试 100 次还是同一个错。把这类异常留在不重试的名单里,让它早点暴露,比藏着强。
7. 参数速查
三个旋钮加起来常用的就这几个,对照着选就够日常用了:
| 参数 | 作用 | 解决的问题 |
|---|---|---|
| stop_after_attempt(n) | 最多尝试 n 次后放弃 | 失败是永久性的,别无限打下去 |
| stop_after_delay(s) | 从首次调用起超过 s 秒就放弃 | 请求有时间预算,不能无限阻塞 |
| wait_fixed(s) | 每次重试固定等 s 秒 | 让对端缓口气,别连续猛打 |
| wait_random(a, b) | 每次在 a 到 b 秒间随机等待 | 打散多客户端同一秒重试 |
| wait_exponential(…) | 等待时间按指数逐次拉长 | 对端越忙退得越远,避免雪崩 |
| retry_if_exception_type(…) | 只在指定异常上重试 | 业务错误和 bug 不被重试掩盖 |
官方文档在 https://tenacity.readthedocs.io/en/latest/ ,除了这里讲的几个,它还支持根据返回值决定是否重试、重试前后执行回调、记录每次重试日志等更细的用法,需要时再去翻。




冀公网安备13050302001966号