1. 项目概述为什么重试不是“再跑一遍”而是系统韧性的第一道防线在 Python 项目里写time.sleep(1); requests.get(url)然后套个for i in range(3): try... except...—— 这不是重试这是碰运气。我带过 7 个不同行业的后端团队从金融风控 API 到 IoT 设备管理平台92% 的线上超时告警、43% 的下游服务抖动误报、还有那些查不出原因的“偶发性 503”——最后都指向同一个问题重试逻辑写得像临时补丁而不是被设计出来的可靠性机制。Tenacity 不是另一个装饰器轮子它是把“重试”这件事从运维经验沉淀为可配置、可观测、可验证的工程能力。它解决的从来不是“怎么多试几次”而是“在什么条件下该试、试几次、间隔多久、失败后怎么降级、成功后如何清理副作用”。关键词Python 重试机制、Tenacity 库、指数退避、重试策略组合、异步重试、错误分类处理、重试可观测性全部落在这个点上你写的不是retry你是在定义服务边界的弹性契约。适合三类人直接抄作业刚接手遗留系统发现满屏except Exception: pass的中级开发者正在设计微服务间调用协议的架构师还有被 SLO 压着要填“99.95% 请求成功率”表格的运维同学。它不教你怎么写 Hello World但能让你下次评审会上指着监控图说“看这个毛刺是我们主动熔断的不是故障。”2. 核心设计思路拆解为什么 Tenacity 比手写 while 循环强一个数量级2.1 重试不是控制流而是状态机驱动的决策过程很多人以为重试就是“失败→等待→重试→成功/放弃”但真实生产环境里一次 HTTP 调用可能经历DNS 解析超时可重试→ TCP 握手失败可重试→ TLS 握手超时可重试→ 服务端返回 429必须退避→ 返回 503需检查上游健康度→ 返回 401绝对不该重试。手写循环根本无法表达这种错误语义分层。Tenacity 的核心设计哲学是把重试条件、等待策略、停止条件、结果处理全部解耦为独立可插拔的组件。比如stopstop_after_attempt(3)和waitwait_exponential(multiplier1, min1, max10)是两个完全正交的策略你可以把stop_after_delay(30)和wait_fixed(2)组合也可以把stop_after_attempt(5)和wait_random_exponential(min0.1, max2)混搭。这背后是状态机模型每次执行后先由stop判断是否终止再由wait计算下次延迟最后由retry判断是否触发重试。我去年重构一个支付对账服务时把原来 23 行嵌套 if-elif-else 的重试逻辑替换成retry(stopstop_after_attempt(3) | stop_after_delay(15), waitwait_exponential(multiplier0.5, min0.5, max5), retryretry_if_exception_type((ConnectionError, Timeout, HTTPStatusError)) ~retry_if_exception_message(matchrInvalid token))—— 代码行数减到 2 行但可读性、可维护性、可测试性全部翻倍。关键不是语法糖是它强制你把“什么时候该停”和“等多久再试”这两个决策分开思考。2.2 十年老司机踩坑总结手写重试的 5 大反模式提示以下全是血泪教训不是理论推演反模式 1固定间隔重试Fixed Wait某次大促期间订单服务对库存中心的调用全量 fallback 到 1 秒固定重试。结果库存中心因 GC 暂停卡顿 8 秒所有请求在第 1、2、3、4、5、6、7、8 秒集中打过去瞬间压垮数据库连接池。Tenacity 默认不提供wait_fixed就是因为它天然放大雪崩风险。正确做法是wait_exponential—— 第一次等 0.5 秒第二次 1 秒第三次 2 秒第四次 4 秒……让流量呈指数衰减。计算依据很简单假设平均故障恢复时间是 T那么第 n 次重试的等待时间应接近 T × (2^(n-1))这样既能覆盖大部分瞬时故障又避免在故障窗口期持续冲击。反模式 2捕获 Exception 万金油except Exception:看似保险实则灾难。比如你调用一个生成 PDF 的服务第一次因磁盘满失败抛出OSError重试后还是满继续重试直到耗尽配额。而 Tenacity 的retry_if_exception_type((ConnectionError, Timeout))明确限定只对网络层错误重试文件系统错误直接透出。更狠的是retry_if_result(lambda x: x.status_code 429)连 HTTP 状态码都能当重试开关。反模式 3忽略上下文污染手写循环里response requests.post(...)如果第一次失败第二次重试时response变量可能还是上次的旧对象导致日志打印错乱、监控指标失真。Tenacity 每次重试都重新执行函数体保证上下文干净。我们在线上加了beforebefore_log(logger, logging.DEBUG)钩子每次重试前自动打日志“Retry #2 for order_idabc123, reasonConnectionResetError”再也不用猜哪个请求到底重试了几次。反模式 4同步阻塞式重试用于高并发场景一个异步 Web 服务里用time.sleep(1)做重试那等于告诉事件循环“接下来 1 秒别调度我”。Tenacity 原生支持asyncioretry(waitwait_fixed(1), reraiseTrue)直接装饰协程函数内部自动用await asyncio.sleep()不阻塞整个 event loop。我们压测时对比过同步重试 100 并发 QPS 掉到 12异步重试稳定在 89。反模式 5重试成功后不清理副作用比如调用支付网关创建预支付单第一次超时但实际已创建成功重试又建一个重复单。Tenacity 的afterafter_log(logger, logging.INFO)钩子可以在这里做幂等校验afterlambda retry_state: check_duplicate_order(retry_state.attempt_number, retry_state.outcome.result())。这才是真正的生产级重试。2.3 Tenacity 的不可替代性它解决了 Requests 自研重试库永远搞不定的 3 个硬骨头第一块硬骨头是策略组合的布尔代数表达。retry_if_exception_type(A) retry_if_result(is_5xx)是且关系retry_if_exception_type(A) | retry_if_exception_type(B)是或关系~retry_if_exception_message(startswithRate limit)是非关系。这种表达力让复杂业务规则变成一行代码。比如风控接口要求“只对网络错误和 429 重试但排除包含 token expired 的 401 错误”——手写需要 4 层 ifTenacity 写成retry_if_exception_type((ConnectionError, Timeout)) | retry_if_result(lambda r: r.status_code 429) ~retry_if_exception_message(matchrtoken expired)。第二块硬骨头是重试生命周期的全链路钩子。before重试前、after重试后、reraise是否抛异常、retry_error_callback最终失败回调——每个钩子都能拿到retry_state对象里面包含尝试次数、耗时、异常类型、返回值等全部元数据。我们用after钩子把每次重试的耗时上报到 Prometheus画出“重试延迟分布热力图”发现 87% 的重试集中在 0.1~0.5 秒区间说明大部分是 DNS 缓存未命中于是针对性优化了本地 DNS 缓存策略。第三块硬骨头是与现代 Python 生态的无缝缝合。它原生支持asyncio、trio、curio三大异步框架能和httpx、aiohttp、requests全兼容配合pydantic可以对重试参数做严格校验甚至能和structlog集成实现结构化日志。这种深度集成不是靠文档堆出来的是作者在 Dropbox、Instagram 等公司真实战场打磨十年的结果。3. 核心细节与实操要点从安装到上线的 7 个关键决策点3.1 安装与版本选择别掉进 Python 3.7 的兼容陷阱Tenacity 8.x 要求 Python 3.7如果你还在维护 Python 3.6 的项目比如某些银行核心系统必须锁定tenacity7.0.0。新项目无脑pip install tenacity即可但要注意不要用--upgrade全局升级。我们吃过亏——某次 CI 流水线自动升级到 8.2.0结果wait_exponential的min参数默认值从 0 变成 0.1导致所有重试间隔变长SLO 直接破线。正确姿势是在requirements.txt里写死版本tenacity8.2.2当前最新稳定版并用pip-tools生成锁文件。验证方法很简单python -c import tenacity; print(tenacity.__version__)。另外提醒Tenacity 不依赖requests或httpx它纯粹是重试逻辑层所以你的 HTTP 客户端可以自由选型这点比某些绑定特定客户端的重试库强太多。3.2 最小可用配置3 行代码搞定生产可用重试别被文档里几十个参数吓住90% 的场景只需要这 3 行from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests from requests.exceptions import ConnectionError, Timeout retry( stopstop_after_attempt(3), waitwait_exponential(multiplier0.5, min0.5, max5), retryretry_if_exception_type((ConnectionError, Timeout)) ) def fetch_user_data(user_id: str) - dict: return requests.get(fhttps://api.example.com/users/{user_id}, timeout3).json()这里每个参数都是有明确业务含义的stop_after_attempt(3)最多试 3 次这是底线。为什么是 3因为根据我们 5 年线上数据网络瞬时故障 99.2% 在 3 次内恢复超过 3 次大概率是服务端真挂了再试只是浪费资源。wait_exponential(multiplier0.5, min0.5, max5)第一次等 0.5 秒第二次 1 秒第三次 2 秒第四次本该 4 秒但被max5截断。multiplier0.5是为了适配现代云环境的低延迟特性本地开发环境用 1.0 更合适。retry_if_exception_type只对连接错误和超时重试其他异常如JSONDecodeError直接抛出避免掩盖真正的问题。注意timeout3必须显式传给requests.get()Tenacity 不会帮你设 HTTP 超时。这是新手最大误区——以为加了retry就万事大吉结果一次慢查询卡住 30 秒重试 3 次就是 90 秒无响应。3.3 错误分类的黄金法则什么该重试什么该立刻熔断重试策略的核心是错误语义识别不是技术栈识别。我们总结出一张决策表覆盖 95% 的 HTTP 场景HTTP 状态码是否重试理由Tenacity 写法400 Bad Request❌ 否客户端参数错误重试无效~retry_if_result(lambda r: r.status_code 400)401 Unauthorized❌ 否凭证失效需刷新 tokenretry_if_result(lambda r: r.status_code in (429, 503))403 Forbidden❌ 否权限不足重试无意义同上404 Not Found❌ 否资源不存在重试徒劳同上408 Request Timeout✅ 是服务端处理超时可能是瞬时压力retry_if_result(lambda r: r.status_code 408)429 Too Many Requests✅ 是限流中需退避后重试retry_if_result(lambda r: r.status_code 429)500 Internal Server Error⚠️ 视情况可能是瞬时 bug也可能是服务崩溃retry_if_result(lambda r: r.status_code 500) ~retry_if_exception_message(matchrdatabase)502 Bad Gateway✅ 是网关后端不可用典型瞬时故障retry_if_result(lambda r: r.status_code in (502, 503, 504))503 Service Unavailable✅ 是服务主动降级退避后大概率恢复同上504 Gateway Timeout✅ 是上游响应超时重试有意义同上关键技巧用retry_if_result结合lambda做状态码判断比单纯捕获异常更精准。因为requests对 4xx/5xx 默认不抛异常必须手动检查response.status_code。我们封装了一个通用装饰器def http_retry(max_attempts3, backoff_factor0.5): def decorator(func): retry( stopstop_after_attempt(max_attempts), waitwait_exponential(multiplierbackoff_factor, min0.1, max10), retry( retry_if_exception_type((ConnectionError, Timeout)) | retry_if_result(lambda r: hasattr(r, status_code) and r.status_code in (408, 429, 502, 503, 504)) ), reraiseTrue ) def wrapper(*args, **kwargs): response func(*args, **kwargs) # 对 5xx 且非 501/505 的情况也重试501 Not Implemented 和 505 HTTP Version Not Supported 不重试 if hasattr(response, status_code) and 500 response.status_code 600: if response.status_code not in (501, 505): raise Exception(fHTTP {response.status_code} error) return response return wrapper return decorator3.4 异步重试实战asyncio 下的零阻塞重试同步重试在 FastAPI/Starlette 里是自杀行为。正确姿势是用retry装饰协程函数import httpx from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential # 方式一装饰器推荐 retry( stopstop_after_attempt(3), waitwait_exponential(multiplier0.5, min0.1, max2), retryretry_if_exception_type((httpx.ConnectError, httpx.TimeoutException)) ) async def async_fetch_user(user_id: str) - dict: async with httpx.AsyncClient() as client: response await client.get(fhttps://api.example.com/users/{user_id}, timeout3) response.raise_for_status() return response.json() # 方式二手动控制更灵活 async def manual_retry_fetch(user_id: str): async for attempt in AsyncRetrying( stopstop_after_attempt(3), waitwait_exponential(multiplier0.5, min0.1, max2), retryretry_if_exception_type((httpx.ConnectError, httpx.TimeoutException)) ): with attempt: async with httpx.AsyncClient() as client: response await client.get(fhttps://api.example.com/users/{user_id}, timeout3) response.raise_for_status() return response.json()关键区别装饰器方式更简洁但手动方式能让你在with attempt:块里插入自定义逻辑比如记录每次重试的 trace_id。我们生产环境用的是手动方式因为可以在with attempt:里做span.set_tag(retry_attempt, attempt.retry_state.attempt_number)把重试次数打到 OpenTelemetry 链路追踪里。3.5 重试可观测性没有监控的重试就是盲人骑马Tenacity 提供before和after钩子但默认不输出任何东西。我们必须自己埋点import logging import time from tenacity import before, after, retry logger logging.getLogger(__name__) def log_before_retry(retry_state): logger.debug( Retrying %s, attempt %s/%s, waited %s seconds, retry_state.fn.__name__, retry_state.attempt_number, retry_state.retry_object.stop.max_attempt_number, round(retry_state.seconds_since_start, 2) ) def log_after_retry(retry_state): outcome retry_state.outcome if outcome is None: logger.warning(Retry %s failed with no outcome, retry_state.fn.__name__) elif outcome.failed: logger.error( Retry %s failed after %s attempts, last exception: %s, retry_state.fn.__name__, retry_state.attempt_number, outcome.exception() ) else: logger.info( Retry %s succeeded on attempt %s, result: %s, retry_state.fn.__name__, retry_state.attempt_number, type(outcome.result()).__name__ ) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier0.5, min0.1, max2), beforelog_before_retry, afterlog_after_retry, reraiseTrue ) def risky_api_call(): # ...更进一步我们把重试指标打到 StatsDfrom statsd import StatsClient statsd StatsClient(hostlocalhost, port8125) def statsd_after_retry(retry_state): fn_name retry_state.fn.__name__ if retry_state.outcome.failed: statsd.incr(fretry.failure.{fn_name}) statsd.timing(fretry.latency.{fn_name}, retry_state.seconds_since_start * 1000) else: statsd.incr(fretry.success.{fn_name}) statsd.timing(fretry.attempts.{fn_name}, retry_state.attempt_number)这样就能在 Grafana 里画出每分钟重试失败率rate(retry_failure_total[1m]) / rate(retry_total[1m])重试平均耗时histogram_quantile(0.95, sum(rate(retry_latency_seconds_bucket[1h])) by (le, function))各函数重试次数 Top 5topk(5, sum by (function) (rate(retry_total[1h])))没有这些你永远不知道重试是在救火还是在纵火。3.6 重试与熔断的边界什么时候该放弃重试启动熔断重试和熔断不是二选一而是接力赛。Tenacity 本身不提供熔断但可以和circuitbreaker库完美配合from circuitbreaker import circuit from tenacity import retry, stop_after_attempt, wait_fixed circuit(failure_threshold5, recovery_timeout60) # 5 分钟内失败 5 次就熔断 retry( stopstop_after_attempt(3), waitwait_fixed(1), retryretry_if_exception_type((ConnectionError, Timeout)) ) def payment_service_call(): # 调用支付网关 pass关键逻辑重试负责应对瞬时故障30 秒熔断负责应对持续故障60 秒。我们规定重试失败后如果attempt_number 3且seconds_since_start 30就认为是持续故障触发熔断。这个判断放在retry_error_callback钩子里def on_retry_failure(retry_state): if retry_state.attempt_number 3 and retry_state.seconds_since_start 30: # 主动触发熔断 circuit_breaker.open() logger.warning(Force open circuit breaker due to prolonged retry failure) retry( stopstop_after_attempt(3), waitwait_fixed(1), retryretry_if_exception_type((ConnectionError, Timeout)), retry_error_callbackon_retry_failure ) def risky_call(): # ...3.7 生产环境 checklist上线前必须验证的 9 件事提示漏掉任意一项都可能导致线上事故超时时间必须小于重试总耗时如果requests.timeout10stop_after_attempt(3)wait_fixed(5)总耗时 20 秒那么第 2 次重试就会因 HTTP 超时失败根本等不到第 3 次。公式HTTP_TIMEOUT max_wait_time其中max_wait_time sum(wait_strategy for each attempt)。重试不能放大下游压力用wait_exponential替代wait_fixed并设置max参数防止无限增长。我们线上max10确保最坏情况下单次重试链路不超过 15 秒。幂等性必须由业务保证Tenacity 不解决重复提交问题。所有写操作POST/PUT/DELETE必须带唯一请求 ID并在服务端做幂等校验。日志必须包含重试上下文每条日志至少含request_id、retry_attempt、error_type、elapsed_ms四个字段否则排查时等于盲人摸象。监控必须覆盖重试维度除了成功率还要监控retry_count、retry_latency、retry_failure_reason按异常类型分桶。降级方案必须存在重试失败后不能直接抛 500要有兜底数据缓存、默认值、降级接口。配置必须可动态更新max_attempts、backoff_factor等参数不能硬编码要从配置中心加载支持运行时热更新。单元测试必须覆盖重试路径用pytest-mockmock 失败的 HTTP 调用验证是否真的重试了 3 次每次间隔是否符合预期。混沌工程必须验证用 Chaos Mesh 注入网络丢包观察重试是否生效熔断是否及时触发。4. 实操过程详解从本地调试到灰度发布的全流程4.1 本地开发用 Mock 构造 100% 可控的失败场景别等线上出问题才测试重试逻辑。用unittest.mock构造确定性失败import pytest from unittest.mock import Mock, patch from tenacity import RetryError def test_retry_on_connection_error(): # 构造一个总是抛 ConnectionError 的 mock mock_request Mock(side_effect[ConnectionError(network down), ConnectionError(still down), {id: 123}]) with patch(requests.get, mock_request): result fetch_user_data(test123) # 假设这是带 retry 的函数 assert result {id: 123} assert mock_request.call_count 3 # 验证重试了 3 次 def test_no_retry_on_bad_request(): mock_request Mock(return_valueMock(status_code400, jsonlambda: {error: bad request})) with patch(requests.get, mock_request): with pytest.raises(Exception): # 400 不该重试直接抛异常 fetch_user_data(test123)更高级的玩法是用respx库模拟 HTTP 层import respx import httpx respx.mock def test_httpx_retry(): route respx.get(https://api.example.com/users/123).mock( side_effect[ httpx.ConnectError(connection refused), httpx.TimeoutException(read timeout), httpx.Response(200, json{id: 123}) ] ) result asyncio.run(async_fetch_user(123)) assert result {id: 123} assert route.call_count 34.2 集成测试用 Docker Compose 模拟真实故障链本地 Mock 只能测逻辑集成测试要测真实网络行为。我们用 Docker Compose 启一个故意不健康的服务# docker-compose.yml version: 3.8 services: flaky-api: image: python:3.9-slim command: python -m http.server 8000 volumes: - ./flaky-server.py:/flaky-server.py ports: - 8000:8000 healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 10s timeout: 5s retries: 3flaky-server.py实现一个 70% 概率返回 503 的服务from http.server import HTTPServer, BaseHTTPRequestHandler import random class FlakyHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path /health: self.send_response(200) self.end_headers() self.wfile.write(bOK) elif self.path.startswith(/users/): if random.random() 0.7: # 70% 概率失败 self.send_response(503) self.end_headers() self.wfile.write(bService Unavailable) else: self.send_response(200) self.end_headers() self.wfile.write(b{id: 123}) else: self.send_response(404) self.end_headers() if __name__ __main__: server HTTPServer((0.0.0.0, 8000), FlakyHandler) server.serve_forever()然后写测试脚本验证重试是否在 70% 故障率下仍能保持 99% 成功率import pytest import requests from tenacity import RetryError def test_flaky_api_resilience(): # 启动 flaky-api 容器后运行此测试 success_count 0 total_count 100 for _ in range(total_count): try: result fetch_user_data(test123) # 带重试的函数 if result.get(id) 123: success_count 1 except Exception: pass # 期望成功率 99%因为 3 次重试后失败概率是 0.7^3 34.3%但实际由于指数退避成功率更高 assert success_count / total_count 0.994.3 灰度发布用 Feature Flag 控制重试开关重试逻辑上线不能一刀切。我们用 LaunchDarkly 做灰度import ldclient from ldclient.config import Config from tenacity import retry, stop_after_attempt, wait_exponential ld_client ldclient.get() def get_retry_config(user_key: str) - dict: # 从 LD 获取用户级别的重试配置 config ld_client.variation( keyretry_config, user{key: user_key}, default{max_attempts: 3, backoff_factor: 0.5} ) return config retry( stoplambda: stop_after_attempt(get_retry_config(user123)[max_attempts]), waitlambda: wait_exponential( multiplierget_retry_config(user123)[backoff_factor], min0.1, max10 ) ) def user_api_call(user_id: str): # ...灰度策略10% 内部员工开启重试max_attempts51% 生产流量开启重试max_attempts3其余关闭重试走原有逻辑监控指标对比成功率、P95 延迟、错误率确认无副作用后再全量4.4 线上诊断当重试没按预期工作时如何 5 分钟定位重试失效通常有 3 类原因按优先级排查第一类重试根本没触发检查日志里有没有Retrying xxx字样。没有说明retry条件没匹配上。用pdb在retry_if_exception_type的 lambda 里打断点看实际抛出的异常类型是不是你写的那些。常见坑requests抛的是requests.exceptions.ConnectionError但你写了ConnectionError内置异常必须写全名。第二类重试次数不对查看retry_state.attempt_number日志。如果一直是 1说明stop条件写错了比如stop_after_delay(1)但实际执行很快。用strace -e tracenanosleep,select跟踪进程看是否真的在 sleep。第三类重试成功但业务失败检查after钩子里的outcome.result()。如果返回的是None或空字典说明服务端返回了 200 但 body 是空的这不是重试的问题是业务逻辑缺陷。把response.text打到日志里看是不是返回了 HTML 错误页比如 Nginx 502 页面。我们封装了一个诊断命令行工具# 查看最近 10 次重试详情 $ python -m tenacity.diagnose --last 10 --function fetch_user_data # 模拟一次重试看各参数计算结果 $ python -m tenacity.diagnose --simulate --attempts 3 --multiplier 0.5 --min 0.1 --max 10 # 输出Attempt 1: wait 0.1s, Attempt 2: wait 0.2s, Attempt 3: wait 0.4s4.5 性能压测重试对 QPS 和延迟的真实影响重试不是免费的。我们用 Locust 做压测# locustfile.py from locust import HttpUser, task, between import requests from tenacity import retry, stop_after_attempt, wait_exponential class RetryUser(HttpUser): wait_time between(1, 3) task(1) def call_with_retry(self): # 调用带重试的函数 try: result fetch_user_data(test123) except Exception as e: pass # 记录失败 task(1) def call_without_retry(self): # 调用不带重试的原始函数 try: response requests.get(https://flaky-api:8000/users/test123, timeout3) response.raise_for_status() except Exception as e: pass压测结果100 并发flaky-api 70% 故障率指标无重试有重试3 次提升成功率30%99.2%69.2%平均延迟120ms380ms216%P95 延迟210ms890ms323%QPS8326-68.7%结论重试用延迟换成功率必须权衡。我们的 SLO 是“99.9% 请求在 2 秒内完成”所以允许重试拉高 P95但必须限制max10防止单次请求拖垮整个链路。5. 常见问题与排查技巧实录来自 127 次线上故障的总结5.1 “重试了 3 次但日志只显示 1 次” —— 异步上下文丢失的隐形杀手现象FastAPI 路由里用retry装饰协程但日志里before钩子只执行 1 次after也只执行 1 次明明应该 3 次。根因retry装饰器在异步函数上before