1. 项目概述为什么重试不是“再跑一遍”而是系统健壮性的第一道防线在 Python 开发中我见过太多人把“网络请求失败”当成一个临时性小问题随手写个while True:加time.sleep(1)就完事——结果上线三天服务在凌晨两点因为第三方 API 偶发超时而卡死日志里全是重复的ConnectionError监控告警响成一片。直到我接手一个支付回调重试模块才真正意识到重试不是容错的补丁而是分布式系统里最基础、最常被低估的稳定性设计单元。Tenacity这个库的名字直译是“坚韧”但它干的活远比字面更精密它不只帮你“多试几次”而是用可组合、可声明、可审计的方式把“什么时候重试”“重试几次”“间隔怎么变”“哪些错误该重试”“重试失败后怎么兜底”全部结构化。它背后是经过 Netflix、Spotify 等高并发场景长期验证的指数退避Exponential Backoff、抖动Jitter、条件判断retry_if_exception_type、结果校验retry_if_result等一整套工程实践。你不需要成为分布式系统专家也能用好 Tenacity——它把复杂逻辑封装成函数装饰器和上下文管理器一行retry(stopstop_after_attempt(3))就能替代手写的 20 行 while 循环但如果你真想吃透它就得理解它如何与 Python 的异常传播机制、异步事件循环、线程/协程生命周期深度咬合。这篇文章就是我过去三年在微服务网关、数据同步管道、IoT 设备指令下发等真实场景中踩过坑、调过参、压过测后整理出的完整实践手册。适合所有正在写requests.get()却还没加重试逻辑的开发者也适合已经用上tenacity但还在靠猜参数的中级工程师。核心关键词已自然嵌入Python 重试机制、Tenacity 库、指数退避、重试策略设计、网络请求容错、异步重试、重试失败兜底。接下来我们不讲抽象理论直接从真实需求出发一层层拆解 Tenacity 是怎么把“再试一次”这件事变成一门可量化、可调试、可演进的工程手艺。2. 内容整体设计与思路拆解为什么 Tenacity 不是 requests 的插件而是错误处理范式的升级2.1 传统重试方案的三大硬伤手写循环、装饰器黑盒、框架绑定先看一个典型的手写重试import time import requests def fetch_user(user_id): for attempt in range(3): try: resp requests.get(fhttps://api.example.com/users/{user_id}) resp.raise_for_status() return resp.json() except (requests.ConnectionError, requests.Timeout) as e: if attempt 2: raise e time.sleep(2 ** attempt) # 简单指数退避这段代码看似能用但实际交付时会暴露三个致命问题逻辑污染业务代码for循环、try/except、time.sleep和真正的业务逻辑requests.getresp.json()混在一起阅读时要跳着看修改重试策略时得改三处次数、异常类型、休眠逻辑。退避策略粗糙不可控2 ** attempt是纯指数增长第 3 次重试要等 4 秒如果上游服务刚好在第 3.5 秒恢复这 4 秒就白白浪费更糟的是所有客户端在同一时刻重试会形成“重试风暴”把刚恢复的服务再次打挂。无法适配异步场景time.sleep()会阻塞整个线程在asyncio或trio环境下必须换成await asyncio.sleep()但手写逻辑又要重来一遍且难以保证await在异常路径下被正确调用。再看早期流行的装饰器方案比如自己封装一个retrydef retry(max_attempts3): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): for i in range(max_attempts): try: return func(*args, **kwargs) except Exception: if i max_attempts - 1: raise time.sleep(1) return wrapper return decorator它解决了部分代码复用问题但依然存在硬伤策略不可组合、状态不可观测、错误不可分类。比如你想“只对 ConnectionError 重试但对 404 不重试”这个装饰器就得加参数你想“前两次等 1 秒第三次等 5 秒”就得重写休眠逻辑你想知道“这次重试是第几次、耗时多久、触发了什么异常”它根本不提供钩子。而 Tenacity 的设计哲学是把重试拆解为四个正交维度Stop停止条件重试多少次最长耗时多久是否达到某个时间点Wait等待策略每次重试前等多久是固定间隔、线性增长、指数退避还是带随机抖动Retry触发条件遇到什么异常或返回什么结果时才重试支持按异常类型、异常消息、返回值内容精细过滤。Before/After生命周期钩子重试前记录日志、重试后发送告警、失败后执行降级逻辑。这四个维度像乐高积木你可以自由拼接stopstop_after_attempt(5) stop_after_delay(30)表示“最多试 5 次且总耗时不超过 30 秒”waitwait_exponential(multiplier1, min1, max10) wait_random(0, 2)表示“指数退避基础上叠加 0~2 秒随机抖动”。这种声明式写法让重试策略本身成为可读、可测、可版本化的配置项而不是散落在各处的魔法数字。2.2 Tenacity 的底层机制如何在不侵入业务逻辑的前提下接管控制流Tenacity 的核心不是魔法而是对 Python 异常处理机制的深度利用。它的retry装饰器本质是一个重试上下文管理器的语法糖内部通过retrying类维护一个状态机关键流程如下首次执行调用被装饰函数若成功则直接返回结果若抛出异常则进入重试决策流程。策略评估将异常对象传给retry条件函数如retry_if_exception_type(ConnectionError)返回True则继续False则立即抛出原异常。等待计算根据当前重试次数、历史耗时等调用wait函数计算本次需休眠的秒数支持浮点数精确到毫秒。休眠执行同步模式下调time.sleep(wait_seconds)异步模式下自动识别async def函数调用await asyncio.sleep(wait_seconds)。循环迭代回到步骤 1直到满足stop条件如达到最大次数或某次执行成功。这个流程的关键在于Tenacity 从不捕获你不关心的异常。比如你配置了retry_if_exception_type(TimeoutError)当函数抛出ValueError时Tenacity 直接透传不会做任何拦截——这保证了业务异常语义的完整性。同时它通过functools.wraps完美保留原函数的__name__、__doc__、签名等元信息对 IDE 自动补全、类型检查工具如 mypy完全透明。提示Tenacity 默认使用time.time()计算耗时这意味着它在asyncio环境下仍能准确统计“墙钟时间”wall-clock time而非协程调度时间。这是很多自研异步重试库容易忽略的细节——它们用asyncio.get_event_loop().time()计算但在协程被挂起期间这个时间不会推进导致“等待 1 秒”实际可能拖成 10 秒。2.3 为什么不用 requests.adapters.Retry它和 Tenacity 的根本差异在哪里requests库自带urllib3.util.retry.Retry很多人会疑惑既然 requests 已经有重试为什么还要 Tenacity答案是作用域不同能力边界不同演进路径不同。requests.adapters.Retry是 HTTP 客户端层的重试它只管urllib3底层的连接、读取、重定向错误且只能作用于requests.Session实例。它无法处理业务层逻辑错误比如 API 返回{code: 500, msg: 库存不足}这属于业务异常urllib3根本不认为这是错误。非 HTTP 操作数据库查询、文件读写、本地函数调用失败requests.Retry完全无能为力。复杂条件判断它只能按 HTTP 状态码如status_forcelist(502, 503, 504)重试无法根据 JSON 响应体中的字段值决定是否重试。而 Tenacity 是通用的 Python 函数重试框架它不关心你调用的是requests、psycopg2还是open()只要是个可调用对象就能套上重试策略。更重要的是它的策略是可编程的你可以写一个函数解析响应 JSON当data[status] PENDING时返回True触发重试这在requests.Retry中根本无法实现。实测对比在一个需要轮询订单状态的场景中我们同时测试两种方案requests.Retry配置status_forcelist[200]强制对 200 也重试但无法区分{status:PROCESSING}和{status:SUCCESS}导致成功后还在空转。Tenacityretry_if_result(lambda r: r.get(status) PROCESSING)精准控制平均减少 62% 的无效请求。这就是通用框架和专用组件的本质区别前者解决“怎么做”后者解决“做什么”。3. 核心细节解析与实操要点从零开始构建一个生产级重试策略3.1 安装与基础用法别急着写策略先搞懂 Tenacity 的“最小可行重试”安装只需一行pip install tenacity注意Tenacity 兼容 Python 3.7且无任何外部依赖不像某些重试库依赖aiohttp或tornado这是一个非常关键的工程优势——它不会给你引入意料之外的依赖冲突。最简用法就是给函数加个装饰器from tenacity import retry, stop_after_attempt retry(stopstop_after_attempt(3)) def unreliable_function(): print(Trying...) raise Exception(Something went wrong)运行它你会看到Trying... Trying... Trying... Traceback (most recent call last): ... Exception: Something went wrong三次尝试后原异常被重新抛出。这里的关键是stop_after_attempt(3)——它表示“最多重试 3 次”注意这不是“总共执行 3 次”而是“重试动作最多发生 3 次”所以函数实际会被调用 4 次首次 3 次重试。这个细节在压测时特别重要如果你配置stop_after_attempt(10)却看到日志里有 11 条“Trying...”别慌这是预期行为。注意Tenacity 默认对所有异常都重试。这在开发阶段方便但生产环境必须显式指定retry条件否则KeyboardInterrupt、SystemExit这类致命异常也会被重试导致程序无法被 CtrlC 中断。安全做法是始终配置retryretry_if_exception_type((ConnectionError, TimeoutError))。3.2 四大核心策略详解Stop、Wait、Retry、Reraise 的参数精讲Stop 策略控制重试的“刹车点”Stop 策略决定何时彻底放弃。Tenacity 提供多个内置函数可单独使用也可用与、|或组合函数参数说明典型场景stop_after_attempt(max_attempt_number)最多重试次数不含首次简单任务如发短信试 3 次不成就走人工通道stop_after_delay(max_delay)总耗时上限秒实时性要求高的接口如支付确认3 秒内必须返回stop_when_event_set(event)监听 threading.Event 信号主进程收到 SIGTERM 时通知所有重试任务立即停止stop_never永不放弃慎用IoT 设备心跳上报宁可一直重试也不能断连组合实战一个支付网关调用要求“最多试 5 次且总耗时不超过 15 秒”from tenacity import stop_after_attempt, stop_after_delay retry( stopstop_after_attempt(5) stop_after_delay(15), # 其他策略... ) def call_payment_gateway(): ...这里是逻辑与意味着两个条件必须同时满足才会停止。如果第 3 次重试后总耗时已达 16 秒即使没到 5 次也会立即终止并抛出异常。Wait 策略让重试“呼吸”避免雪崩Wait 策略决定每次重试前的等待时长。没有合理的 Wait重试就是自杀式攻击。Tenacity 的 Wait 策略设计极其精细重点掌握以下三个wait_fixed(wait)固定间隔如wait_fixed(2)每次等 2 秒。适合调试但生产环境慎用——所有客户端同步重试会形成脉冲流量。wait_exponential(multiplier1, min0, max30)指数退避公式为min multiplier * (2 ** attempt)默认min0,max30。例如第 1 次重试等 1 秒第 2 次等 2 秒第 3 次等 4 秒……第 6 次等 32 秒但因max30实际等 30 秒。wait_random_exponential(multiplier1, max60)在指数退避基础上叠加0~max毫秒的随机抖动这是对抗“重试风暴”的黄金策略。抖动原理实测在 100 个并发请求中使用wait_exponential第 3 次重试集中在t7±0.1秒而wait_random_exponential让它们分散在t7±2.5秒流量峰值下降 40%。实操心得永远不要在生产环境用wait_fixed。哪怕只是wait_random(0, 1)0~1 秒随机也比固定值强十倍。抖动不是“锦上添花”而是“雪崩防火墙”。Retry 策略精准识别“值得重试”的错误这是 Tenacity 最强大的能力。retry参数接受一个返回布尔值的函数只有返回True时才重试。常用内置函数retry_if_exception_type(*exception_types)按异常类型过滤最常用。支持元组retry_if_exception_type((ConnectionError, TimeoutError))。retry_if_exception_message(matchtimeout, regexTrue)按异常消息匹配regexTrue时支持正则。retry_if_result(predicate)对函数返回值做判断predicate接收返回值返回True则重试。retry_if_not_result(predicate)与上相反返回False才重试。业务场景实战调用一个订单查询 API返回 JSON其中status字段为PROCESSING时表示还在处理需要重试SUCCESS或FAILED则无需重试import json import requests from tenacity import retry, retry_if_result, stop_after_attempt def is_processing(response): try: data response.json() return data.get(status) PROCESSING except (json.JSONDecodeError, AttributeError): return False retry( retryretry_if_result(is_processing), stopstop_after_attempt(10), waitwait_random_exponential(multiplier1, max10) ) def poll_order_status(order_id): resp requests.get(fhttps://api.example.com/orders/{order_id}) return resp # 返回 Response 对象供 is_processing 解析这里is_processing函数接收requests.Response对象解析 JSON 并判断状态。如果 API 返回非 JSON 或无status字段函数返回FalseTenacity 就不会重试而是直接返回该响应——这比try/except更优雅地处理了“协议不一致”的边界情况。Reraise 策略重试失败后如何优雅地“认输”当所有重试都失败后Tenacity 默认会抛出最后一次的异常。但有时你需要抛出一个包装后的业务异常隐藏底层细节返回一个默认值如None或空字典让上游继续执行记录关键指标后再抛出原异常。Tenacity 通过reraise参数和after钩子实现reraiseTrue默认抛出最后一次异常。reraiseFalse返回最后一次调用的返回值如果有的话但若最后一次也抛异常则仍抛出。更灵活的是after钩子它在每次重试后包括最终失败执行from tenacity import after_log import logging logger logging.getLogger(__name__) retry( stopstop_after_attempt(3), afterafter_log(logger, logging.WARNING) # 每次重试后打 WARNING 日志 ) def risky_operation(): ...after_log是内置钩子它会记录重试次数、耗时、异常类型。你也可以写自己的钩子def after_callback(retry_state): if retry_state.outcome.failed: # 重试失败执行降级逻辑 fallback_result get_from_cache() logger.warning(All retries failed, using fallback: %s, fallback_result) return fallback_result else: logger.info(Retry succeeded on attempt %s, retry_state.attempt_number) retry(afterafter_callback) def operation_with_fallback(): ...retry_state是 Tenacity 内部的状态对象包含attempt_number、outstanding是否还在重试中、outcome成功或失败的结果等丰富信息是编写高级钩子的基础。3.3 同步与异步的无缝切换同一套策略两套执行引擎Tenacity 最惊艳的设计之一是它对同步/异步的“零感知”支持。你写一套策略它能自动适配如果被装饰函数是def同步它用time.sleep()如果被装饰函数是async def异步它用await asyncio.sleep()甚至支持trio、curio等其他异步框架需手动指定loop。同步示例import time from tenacity import retry, wait_fixed retry(waitwait_fixed(1)) def sync_fetch(): time.sleep(0.1) # 模拟慢操作 if time.time() % 5 1: # 20% 概率失败 raise ConnectionError(Network flaky) return success异步示例import asyncio from tenacity import retry, wait_fixed retry(waitwait_fixed(1)) async def async_fetch(): await asyncio.sleep(0.1) if asyncio.get_event_loop().time() % 5 1: raise ConnectionError(Network flaky) return success注意两个函数的装饰器完全一样Tenacity 在运行时通过inspect.iscoroutinefunction()自动识别函数类型并选择对应的休眠方式。这意味着你的重试策略可以跨同步/异步服务复用无需为不同框架写两套逻辑。实操心得在 FastAPI 或 Starlette 项目中我习惯把重试策略定义在config.py里然后在路由函数和后台任务函数中统一引用。这样当需要调整重试参数如把max_delay从 10 秒改成 30 秒时只需改一处配置所有相关接口自动生效。4. 实操过程与核心环节实现从开发到上线的全链路落地指南4.1 构建一个真实的电商库存扣减重试系统我们以一个典型的电商场景为例用户下单时需要调用库存服务扣减商品库存。库存服务偶发超时或 503但业务要求“尽最大努力扣减失败则提示用户稍后再试”。第一步定义库存服务客户端# inventory_client.py import requests from tenacity import ( retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type, before_log, after_log, ) import logging logger logging.getLogger(__name__) class InventoryClient: def __init__(self, base_url: str): self.base_url base_url.rstrip(/) self.session requests.Session() retry( stopstop_after_attempt(5), waitwait_random_exponential(multiplier1, max10), retryretry_if_exception_type(( requests.ConnectionError, requests.Timeout, requests.HTTPError, )), beforebefore_log(logger, logging.DEBUG), afterafter_log(logger, logging.WARNING), reraiseTrue, ) def deduct(self, sku_id: str, quantity: int) - dict: 扣减库存返回 {success: true, locked_quantity: 10} 若返回 503 或网络错误自动重试 url f{self.base_url}/inventory/deduct payload {sku_id: sku_id, quantity: quantity} try: resp self.session.post(url, jsonpayload, timeout(3.05, 15)) resp.raise_for_status() # 对 4xx/5xx 抛出 HTTPError return resp.json() except requests.HTTPError as e: # 特殊处理409 Conflict库存不足不重试直接抛出 if resp.status_code 409: raise e raise这里我们组合了多个策略stop_after_attempt(5)最多重试 5 次wait_random_exponential指数退避 随机抖动防雪崩retry_if_exception_type只对网络错误和 HTTP 错误重试before/after_log记录每次重试的详细日志便于排查关键的except块对409 Conflict库存不足这种业务确定性错误主动raise终止重试避免无意义轮询。第二步集成到下单服务# order_service.py from inventory_client import InventoryClient inventory_client InventoryClient(https://inventory-api.example.com) def create_order(user_id: str, items: list): # ... 其他订单创建逻辑 for item in items: try: result inventory_client.deduct(item[sku_id], item[quantity]) if not result.get(success): raise RuntimeError(fInventory deduct failed: {result}) except requests.HTTPError as e: if e.response.status_code 409: raise InsufficientStockError(fSKU {item[sku_id]} out of stock) else: # 其他 HTTP 错误如 503已被 Tenacity 重试到这里说明重试也失败 raise ServiceUnavailableError(Inventory service unavailable) except Exception as e: # Tenacity 重试后仍失败的异常如 ConnectionError raise ServiceUnavailableError(Inventory service unreachable) # ... 创建订单主表 return {order_id: xxx}第三步添加可观测性——让重试行为可追踪光有重试不够还得知道它“重试了几次、花了多久、为什么失败”。我们在InventoryClient.deduct中加入 Prometheus 指标# metrics.py from prometheus_client import Counter, Histogram # 重试次数统计 RETRY_COUNTER Counter( inventory_deduct_retries_total, Total number of inventory deduct retries, [status, exception_type], ) # 重试耗时分布 RETRY_DURATION Histogram( inventory_deduct_retry_duration_seconds, Inventory deduct retry duration in seconds, buckets[0.1, 0.5, 1, 2, 5, 10, 30], ) # 在 deduct 方法中before 钩子里记录开始 def before_retry(retry_state): RETRY_COUNTER.labels(statusstarted, exception_type).inc() # 在 after 钩子里记录结果 def after_retry(retry_state): if retry_state.outcome.failed: exc retry_state.outcome.exception() exc_type type(exc).__name__ RETRY_COUNTER.labels(statusfailed, exception_typeexc_type).inc() RETRY_DURATION.observe(retry_state.seconds_since_start) else: RETRY_COUNTER.labels(statussuccess, exception_type).inc() RETRY_DURATION.observe(retry_state.seconds_since_start)这样运维同学就能在 Grafana 里看到过去一小时库存扣减重试成功率 99.8%平均重试 1.2 次95% 的重试在 2 秒内完成——这才是真正的“可运维”。4.2 异步场景深度实践在 FastAPI 中实现非阻塞重试FastAPI 默认是异步的但很多开发者会不小心写出“伪异步”代码。比如下面这个错误示范# ❌ 错误在 async 函数里调用同步 requests router.post(/order) async def create_order_async(): # 这里 requests.get 是同步阻塞的会阻塞整个 event loop resp requests.get(https://inventory-api.com/stock) return {stock: resp.json()}正确做法是用httpx.AsyncClient替代requests# inventory_async_client.py import httpx from tenacity import retry, wait_random_exponential, stop_after_attempt, retry_if_exception_type class AsyncInventoryClient: def __init__(self, base_url: str): self.base_url base_url.rstrip(/) self.client httpx.AsyncClient(timeouthttpx.Timeout(3.05, read15)) retry( stopstop_after_attempt(5), waitwait_random_exponential(multiplier1, max10), retryretry_if_exception_type(( httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError, )), ) async def deduct(self, sku_id: str, quantity: int) - dict: url f{self.base_url}/inventory/deduct payload {sku_id: sku_id, quantity: quantity} try: resp await self.client.post(url, jsonpayload) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as e: if e.response.status_code 409: raise e raise注意httpx.AsyncClient的post方法是awaitable 的Tenacity 会自动识别并用await asyncio.sleep()休眠。性能对比实测在 1000 QPS 压测下同步requests版本因阻塞 event loop平均响应时间飙升至 1200ms而异步httpx Tenacity 版本稳定在 85ms吞吐量提升 14 倍。这就是“正确异步”的威力。4.3 生产环境避坑指南那些文档里不会写的血泪教训坑 1全局 Session 复用导致连接泄漏很多人会这样写# ❌ 危险全局 session 在重试中可能被意外关闭 session requests.Session() retry(...) def bad_func(): resp session.get(...) # 如果重试中 session 被 close后续会报错 return respTenacity 的重试是在同一个函数调用栈内进行的但如果session在某次重试中被close()下次重试就会失败。正确做法是每次重试都新建轻量级 client或确保 session 生命周期覆盖整个重试周期。对于httpx.AsyncClient推荐用contextlib.AsyncExitStack管理from contextlib import AsyncExitStack retry(...) async def safe_deduct(sku_id: str): async with AsyncExitStack() as stack: client await stack.enter_async_context(httpx.AsyncClient()) resp await client.post(..., json{sku_id: sku_id}) return resp.json()坑 2日志爆炸——重试 10 次打了 100 行 DEBUG 日志Tenacity 的before_log和after_log默认打DEBUG和WARNING如果重试 10 次日志量翻 10 倍。线上环境建议before_log改用INFO级别只记录“开始重试”after_log保留WARNING但只在失败时打或者自定义钩子聚合重试信息def compact_after(retry_state): if retry_state.outcome.failed: logger.warning( Deduct failed after %d attempts, total time %.2fs, last error: %s, retry_state.attempt_number, retry_state.seconds_since_start, str(retry_state.outcome.exception()), )坑 3异步重试中未 await 导致“静默失败”# ❌ 错误忘记 await函数立即返回一个 coroutine 对象 retry(...) async def async_func(): ... result async_func() # 这里 result 是 coroutine不是实际结果Tenacity 的异步装饰器返回的仍是coroutine你必须await它。IDE如 PyCharm通常会警告“calling async function without await”但 CI 流水线里容易漏掉。最佳实践在单元测试中强制检查返回值类型import pytest import asyncio pytest.mark.asyncio async def test_async_deduct_returns_dict(): result await async_inventory_client.deduct(sku-001, 1) assert isinstance(result, dict) assert success in result5. 常见问题与排查技巧实录来自线上事故的 7 个真实案例5.1 问题速查表症状、原因、解决方案症状可能原因解决方案重试从未触发retry条件函数返回False或异常类型不在retry_if_exception_type列表中用print()或logging.debug()在retry函数里打日志确认输入异常对象和返回值重试次数远超预期stop条件配置错误如stop_after_attempt(1)实际调用 2 次或wait时间太短导致高频重试检查retry_state.attempt_number日志确认实际重试次数用stop_after_delay加兜底异步函数重试后报RuntimeWarning: coroutine xxx was never awaited调用方忘记await被装饰的异步函数在入口函数如 FastAPI 路由中确保所有retry异步函数都被await重试过程中内存持续增长retry装饰器在每次重试时创建新闭包持有大量局部变量引用避免在重试函数内定义大型对象如 pandas DataFrame用del显式释放日志显示重试成功但业务逻辑没生效retry_if_result函数逻辑错误误判了成功状态在retry_if_result函数里打印返回值确认判断逻辑用retry_if_not_result反向验证重试耗时远超wait配置函数本身执行时间很长如数据库慢查询wait只控制“间隔”不控制“执行”用stop_after_delay限制总耗时优化函数内部性能多线程环境下重试行为异常threading.local()变量在重试中被复用导致状态污染避免在重试函数中使用threading.local()改用函数参数传递状态5.2 案例 1重试“成功”了但库存被扣了两次