最近是真的忙，作为研三的科研苦逼，一直被毕业论文的进度推着走。
前几天我在跑一组高分三号（GF-3）SAR数据的预处理脚本。几百个GB的影像，还要调用GDAL做些底层的栅格计算和格式转换，想着处理完就能直接拿去跑土壤水分的反演模型了。为了节省时间，我把脚本挂在教研室的服务器上，满心欢喜地回宿舍睡觉。
结果第二天满怀期待地打开终端一看，心肺骤停。凌晨两点多的时候，因为校园网突然抽风，一个中间件的下载请求超时，整个程序直接报 ConnectionError 崩溃了。跑了好几个小时的心血全白费，进度又要往后推一天。
痛定思痛，我决定把代码里的容错机制彻底翻新一遍。以前写爬虫或者批量下载数据，遇到网络不稳，我都是用 while True 嵌套一堆 try-except，再加上干巴巴的 time.sleep()，代码看起来又臭又长，维护起来极其痛苦。
直到我挖到了 tenacity 这个宝藏级别的Python库，简直是相见恨晚。今天就跟大家盘一盘，怎么用这个极其优雅的重试模块，拯救我们脆弱的代码。
到底什么是 Tenacity？
简单来说，Tenacity 是一个用 Python 写的重试库。它的核心理念就是：把“重试”这个动作从你的业务逻辑里抽离出来。
你不需要再写那些繁琐的循环和异常捕获，只需要在需要重试的函数头上加一个 @retry 装饰器，一切就搞定了。它的逻辑非常清晰，我们可以用一个简单的流程图来理解：
[ 开始调用函数 ] | v( 执行核心业务代码 ) | +-------( 执行成功 ) --------> 返回数据，任务结束。 | +-------( 抛出异常报错 ) | v [ Tenacity 拦截异常 ] | 判断是否满足配置的重试条件？(如最大次数、最长等待时间) / \ (是) (否) / \进入等待策略(固定时间/指数退避等) 终止拦截，向外抛出真实异常 | (程序报错或进入后续兜底逻辑)回到 ( 执行核心业务代码 )
废话不多说，我们直接上代码，看看它到底有多好用。
首先，安装非常简单：
场景一：最暴力的无脑重试
假设我们有一个极其不稳定的函数，时不时就会抛出异常。如果什么参数都不加，Tenacity 会在这个函数报错时，无限期、无间隔地一直重试下去，直到成功为止。
import randomfrom tenacity import retry@retrydef unstable_network_request(): print("正在尝试请求服务器数据...") if random.randint(1, 10) < 8: print("网络波动，请求失败！") raise Exception("服务器没有响应") print("请求成功，获取到关键数据！") return "Data Payload"unstable_network_request()
但这其实很危险，因为如果真的是永久性错误，程序就会陷入死循环。所以我们必须给它加点限制。
场景二：限制重试次数与时间
在实际处理遥感数据或者请求API时，我们通常只会尝试几次，如果还是不行就放弃，或者限定总的重试时间不能超过多久。这时候可以使用 stop_after_attempt 和 stop_after_delay。
from tenacity import retry, stop_after_attempt, stop_after_delay# 最多只重试 3 次，3次全败就认命抛出异常@retry(stop=stop_after_attempt(3))def download_sar_image(): print("尝试下载 SAR 影像文件...") raise ConnectionError("下载中断")# 不管重试几次，总重试时间不能超过 10 秒@retry(stop=stop_after_delay(10))def process_raster_data(): print("正在处理栅格矩阵...") raise ValueError("数据类型不匹配")# 当然，你还可以把它们组合起来用，满足任意一个条件就停止 (使用 | 符号)@retry(stop=(stop_after_attempt(5) | stop_after_delay(15)))def combined_task(): print("重试最多5次，且总耗时不超过15秒...") raise Exception("任务失败")
场景三：更聪明的等待策略 (退避算法)
遇到网络拥堵时，马上疯狂重试不仅没用，还容易被目标服务器直接封IP。科学的做法是“退避重试”，比如第一次失败等2秒，第二次失败等4秒，第三次等8秒。Tenacity 的 wait 模块完美解决了这个问题。
from tenacity import retry, wait_fixed, wait_exponential# 策略1：每次重试之间，死板地固定等待 3 秒@retry(wait=wait_fixed(3))def fixed_wait_task(): print("失败了，我等3秒后再来...") raise Exception("Error")# 策略2：指数退避 (这在爬虫和API调用里最常用了！)# multiplier: 乘数，初值。 max: 最大等待时间边界。# 等待时间大概是这样：2s, 4s, 8s, 10s(达到最大限制), 10s...@retry(wait=wait_exponential(multiplier=2, max=10))def exponential_wait_task(): print("遇到了高并发限流，我要慢慢拉长等待时间...") raise Exception("Rate Limit Exceeded")
场景四：按需重试（不是什么错都值得重试）
写代码最怕的就是掩盖真正的 Bug。如果是因为网络断了报 URLError，重试是有意义的；但如果是因为我代码里写错了一个变量名导致 NameError，你重试一万次也没用啊！
所以，我们可以指定只有遇到特定异常时，才触发重试机制。
import urllib.requestfrom urllib.error import URLErrorfrom tenacity import retry, retry_if_exception_type, stop_after_attempt# 明确指明：只有遇到 URLError 的时候我才重试，其他报错直接停机@retry( retry=retry_if_exception_type(URLError), stop=stop_after_attempt(3))def fetch_api_data(url): print(f"正在抓取：{url}") # 模拟一个会抛出 URLError 的请求 response = urllib.request.urlopen(url) return response.read()# 如果函数里抛出的是 TypeError 或者 ValueError，程序会直接崩溃，不会浪费时间重试
实战演练：一个稳健的下载器
最后，我们把上面的技巧融合在一起，写一个在实际做科研项目时非常实用的文件下载辅助函数。它具备了限制次数、指数退避等待、按特定异常重试三大功能。
import requestsfrom requests.exceptions import Timeout, ConnectionErrorfrom tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type# 终极形态的重试装饰器@retry( # 只捕获超时和连接断开的异常 retry=retry_if_exception_type((Timeout, ConnectionError)), # 最多重试 5 次 stop=stop_after_attempt(5), # 等待时间：1s, 2s, 4s, 8s, 10s... wait=wait_exponential(multiplier=1, max=10))def robust_download(url, save_path): print(f"发起请求: {url}") # 设置一个较短的超时时间，模拟严苛的网络环境 response = requests.get(url, timeout=3) response.raise_for_status() with open(save_path, 'wb') as f: f.write(response.content) print(f"文件完美下载并保存至: {save_path}")# 调用示例# robust_download("http://example.com/massive_data.zip", "./data.zip")
写在最后
有了 tenacity 之后，我的数据处理脚本再也没有出现过半夜因为网络抽风而暴毙的情况。只需要在容易出问题的IO操作、网络请求上优雅地加上一个 @retry 装饰器，主业务逻辑的整洁度直线上升。
如果你平时也需要写爬虫、调用第三方API，或者像我一样经常需要自动化跑批量的文件处理，墙裂建议把这个库加入你的日常工具箱。
如果这篇文章帮到了你，或者让你少掉了几根头发，欢迎点赞和转发，这也是我继续在这个账号更新Python技术干货的最大动力！我们下期见。
10个月宝宝每天需要喝多少奶粉？
