Python API调用崩溃?Tenacity库让你轻松实现智能重试

你有没有遇到过这种情况：线上服务调用别人的API，本来跑得好好的，突然网络抖了一下，直接报错崩溃了。我之前就栽过这个坑——凌晨三点被电话叫醒，服务器报警，说是某个第三方接口超时了。结果排查了半天，发现就是网络抖动导致的临时失败，重试一下就好了。

这种事遇多了，我就开始琢磨，能不能写个优雅的重试逻辑？答案就是今天要介绍的这个库——Tenacity。

说实话，我第一次看到Tenacity的时候，脑子里就一个想法：Python圈真是卷得厉害，连重试这种"小功能"都能做得这么精致。这个库让我重新认识了什么叫"好的代码不需要解释"。

 01. 

一、安装与快速上手

Tenacity的安装非常简单，直接pip搞定：

BASHpip install tenacity

对了，当前最新版本是9.1.4，支持Python 3.10+。如果你还在用Python 3.9，需要升级一下了，官方已经停止对3.9的支持。

最基础的使用方式，就是给可能出问题的函数加个@retry装饰器：

PYTHONfrom tenacity import retryimport random@retrydef call_api():    if random.random() < 0.7:  # 70%概率失败        raise Exception("网络请求失败")    return "成功"

这种写法的好处是什么？业务逻辑和重试逻辑完全分离。你不用在函数里写while循环，不用自己维护计数器，重试的事交给Tenacity就好。

我第一次用这个装饰器的时候，感觉就像找到了救命稻草。以前那些丑陋的while循环，终于可以退休了。

 02. 

二、停止条件：什么时候该放弃

默认情况下，@retry会无限重试，直到成功为止。这在测试环境没问题，但生产环境可不敢这么玩——无限重试意味着可能把服务器拖死。

Tenacity提供了几种停止策略，先看最常用的两个。

限制重试次数

PYTHONfrom tenacity import retry, stop_after_attempt@retry(stop=stop_after_attempt(7))def call_api():    # 最多尝试7次，7次都失败就放弃    raise Exception("API调用失败")

这个是我最常用的策略。设置一个合理的重试次数上限，既给了对方服务"恢复"的机会，又不会让自己陷入无尽的等待。

限制总耗时

PYTHONfrom tenacity import retry, stop_after_delay@retry(stop=stop_after_delay(30))def call_api():    # 30秒内不断重试，超时就放弃    raise Exception("API调用失败")

这个策略适合那些对时间敏感的场景。比如用户在前端等着你返回数据，总不能让人家等一分钟吧。

组合条件

最实用的写法是组合多个条件：

PYTHONfrom tenacity import retry, stop_after_attempt, stop_after_delay@retry(stop=(stop_after_delay(30) | stop_after_attempt(5)))def call_api():    raise Exception("API调用失败")

这行代码的意思是：要么时间超过30秒，要么重试次数超过5次，哪个先到就放弃。我管这个叫"双重保险"策略，实际项目中用得最多的就是它。

 03. 

三、等待策略：两次重试之间等多久

光有停止条件还不够，重试间隔也很关键。如果失败后立刻重试，很可能还没等对方恢复呢，你又撞上去了。

固定间隔

PYTHONfrom tenacity import retry, wait_fixed@retry(wait=wait_fixed(2))def call_api():    # 每次失败后等2秒再重试    raise Exception("API调用失败")

这个适合那些"稍等一下就好"的场景。比如对方服务正在重启，等个几秒就好了。

指数退避

固定间隔有个问题——如果对方需要更长的恢复时间呢？指数退避就是来解决这个的：

PYTHONfrom tenacity import retry, wait_exponential@retry(wait=wait_exponential(multiplier=1, min=2, max=60))def call_api():    # 等待时间：2s, 4s, 8s, 16s, 32s, 60s(封顶)    raise Exception("API调用失败")

这里multiplier=1表示倍数是2的指数次方，min=2是最少等2秒，max=60是最多等60秒。

我第一次用指数退避的时候，感觉特别有安全感。随着重试次数增加，等待时间变长，给对方服务足够的时间去恢复。但后来踩了一个坑——有些场景不适合指数退避，后面会说到。

随机抖动

指数退避还有一个问题：如果所有客户端都是同步重试（比如同时部署、同时重启），会造成"惊群效应"——大家同时重试，同时失败，又同时重试。

解决办法是加入随机抖动：

PYTHONfrom tenacity import retry, wait_random@retry(wait=wait_random(min=1, max=3))def call_api():    # 等待1-3秒之间的随机时间    raise Exception("API调用失败")

或者把指数退避和抖动结合起来：

PYTHONfrom tenacity import retry, wait_exponential, wait_random@retry(wait=wait_exponential(multiplier=1, max=60) + wait_random(0, 2))def call_api():    # 指数退避 + 随机抖动    raise Exception("API调用失败")

重要提醒：抖动不是万能的。如果你用的指数退避策略，等待时间已经足够长，就没必要再加抖动了。我之前就见过有人同时用wait_exponential和wait_random，结果抖动把指数退避的效果给抵消了，等于白等那么久。

进阶：等待链

有些场景下，我们需要分阶段采用不同的等待策略：

PYTHONfrom tenacity import retry, wait_chain@retry(wait=wait_chain(    *[wait_fixed(1) for _ in range(3)],      # 前3次重试：等1秒    *[wait_fixed(5) for _ in range(5)],      # 接下来5次重试：等5秒    wait_fixed(10)                             # 之后全部：等10秒))def call_api():    raise Exception("API调用失败")

这种写法适合那种"先快速尝试，不行再慢慢等"的场景。

 04. 

四、异常过滤：只对特定的错误重试

不是所有异常都需要重试。有些错误是"不可恢复"的，比如参数错误、权限不足，这种时候重试一万次也不会有结果。

根据异常类型过滤

PYTHONfrom tenacity import retry, retry_if_exception_type@retry(retry=retry_if_exception_type(IOError))def might_io_error():    # 只对IOError重试，其他异常直接抛出    raise Exception("业务逻辑错误")@retry(retry=retry_if_exception_type((ConnectionError, TimeoutError)))def might_timeout():    # 对网络相关错误重试    raise Exception("超时")

根据异常内容过滤

有时候异常类型不够精确，需要根据异常信息来判断：

PYTHONfrom tenacity import retry, retry_if_exception_message@retry(retry=retry_if_exception_message(match="rate limit"))def call_api():    # 只有错误信息包含"rate limit"才重试    raise Exception("API rate limit exceeded")

组合条件

多个条件可以组合使用：

PYTHONfrom tenacity import retry, retry_if_exception_type, retry_if_resultdef is_error_result(result):    return result is None or result.get("code") != 0@retry(retry=(retry_if_exception_type(IOError) | retry_if_result(is_error_result)))def call_api():    # 网络错误或返回错误结果，都重试    pass

我踩过的最大的坑，就是这里了。有一段时间，我给所有异常都加了重试，结果某些业务错误也被重试了，白白浪费了资源。教训是：重试前一定要想清楚，哪些错误值得重试。

 05. 

五、日志与回调：让重试"可见"

重试逻辑一旦加上，你可能想知道：什么时候触发了重试？重试了几次？为什么失败？

Tenacity提供了完善的日志和回调机制。

日志记录

PYTHONimport loggingfrom tenacity import retry, stop_after_attempt, before_sleep_loglogger = logging.getLogger(__name__)@retry(    stop=stop_after_attempt(3),    before_sleep=before_sleep_log(logger, logging.WARNING))def call_api():    raise Exception("API调用失败")

运行效果：

CODEWARNING - Retrying call_api in 2.0 seconds as it raised Exception('API调用失败').WARNING - Retrying call_api in 4.0 seconds as it raised Exception('API调用失败').

这对于排查线上问题特别有用。我现在养成了习惯，所有重试都要加日志，不然出问题都不知道重试了多少次。

自定义回调

除了日志，你还可以定义更复杂的回调：

PYTHONfrom tenacity import retry, stop_after_attempt, RetryCallStatedef my_callback(retry_state: RetryCallState):    print(f"重试次数: {retry_state.attempt_number}")    print(f"异常: {retry_state.outcome.exception()}")    print(f"耗时: {retry_state.seconds_since_start:.2f}秒")@retry(stop=stop_after_attempt(3), before_sleep=my_callback)def call_api():    raise Exception("API调用失败")

生产环境中，我见过有人用回调来发送告警短信、重试次数统计、甚至自动扩容。这些都是后话了，但Tenacity确实支持这些扩展。

 06. 

六、代码块重试：不需要包装成函数

有时候你的代码逻辑比较复杂，不方便包装成独立函数。Tenacity也考虑到了这一点，提供了上下文管理器的方式：

PYTHONfrom tenacity import Retrying, RetryError, stop_after_attemptdef process_data():    for attempt in Retrying(stop=stop_after_attempt(3)):        with attempt:            # 这里的代码会自动重试            result = call_api()            validate_result(result)

这种方式有个好处：可以在重试块内部访问上下文变量。比如你想在重试时保留前一次的结果做对比，用装饰器就不太好实现，但用上下文管理器就很自然。

 07. 

七、异步支持：让重试不阻塞

现在异步编程越来越普及了，Tenacity也原生支持异步：

PYTHONfrom tenacity import retry, stop_after_attemptimport asyncio@retry(stop=stop_after_attempt(3))async def call_api_async():    # 异步API调用    response = await httpx.AsyncClient().get("https://api.example.com")    return response.json()

异步版本的好处是：重试期间的等待不会阻塞事件循环。如果你用同步重试，在等待的那几秒里，整个事件循环都卡住了；但用异步重试，其他任务照常运行。

Tenacity还支持Tornado和Trio：

PYTHON# Tornado@retry@tornado.gen.coroutinedef tornado_operation():    yield http_client.fetch(url)# Trio（需要传sleep参数）@retry(sleep=trio.sleep)async def trio_operation():    await trio.socket.getaddrinfo('8.8.8.8', 53)

 08. 

八、生产环境避坑指南

说了这么多，最后聊聊实际生产中容易踩的坑。

第一坑：无限重试。默认配置是无限重试，这是最危险的配置。我见过有人在生产环境用了@retry没加停止条件，结果第三方服务挂了，整个系统陷入疯狂重试，CPU跑满。

第二坑：重试风暴。当多个服务实例同时重启或网络抖动时，会出现大量重试请求打到同一个服务。解决方法是加入随机抖动，或者用指数退避+封顶值。

第三坑：幂等性。不是所有接口都支持重试。有些接口是非幂等的，多次调用会产生副作用。比如扣款接口，重试可能导致重复扣款。重要的事说三遍：重试前一定要确认接口幂等性！

第四坑：异常信息泄露。Tenacity默认会把完整的堆栈信息记录下来，这在生产环境可能造成问题——敏感信息可能被日志系统收集。我一般会自定义回调，把敏感字段脱敏后再记录。

 09. 

九、总结

Tenacity这个库，说大不大，说小不小。它的功能很专注——就是帮你处理重试逻辑。但就是这样一个"小功能"，它做得足够深入、足够优雅。

从基础的装饰器语法，到指数退避、随机抖动，再到异步支持、回调机制，Tenacity几乎覆盖了所有重试场景。而且它的API设计得非常直观，看一遍文档就能上手。

如果你还在用自己写的while循环做重试，真的建议试试Tenacity。代码更简洁，逻辑更清晰，出问题也好排查。

当然，工具再好也只是工具。重试策略的设计才是核心。什么时候重试、重试几次、间隔多久，这些决策需要根据具体的业务场景和第三方服务的特点来定。Tenacity只是帮你更优雅地实现这些策略。

好了，今天的分享就到这里。我是Python与AI智能研习社的小编，如果你觉得有用，欢迎转发给需要的朋友。

下期见！