写爬虫调第三方API，总有接口偶尔抽风，返回个超时或者500。最简单的办法是在外面套一层循环，失败了就再来一次。但你把重试逻辑到处复制，过两天想改个等待间隔，就得全局搜索一遍。后来我干脆写了个装饰器，专门干指数退避重试这活，代码清爽多了，也不怕接口随机抖动了。
今天我就把这个装饰器的实现过程拆开讲一遍，从最笨的版本开始，一步一步加功能，最终变成能同步异步通吃的重试工具。
最常见的重试写法，哪儿不好？
for i in range(3):
try:
response = requests.get("https://api.example.com/data")
break
except requests.exceptions.RequestException as e:
print(f"第{i+1}次请求失败：{e}")
time.sleep(2)
else:
raise Exception("重试三次还是挂了")这个写法有俩问题：第一是等待时间固定，如果对方服务一直处于过载状态，你固定等2秒再试，可能还是撞枪口上。第二是这段逻辑没法复用，二十个接口就要复制二十份同样的代码，废手指。
回归本质，重试的核心就两件事：什么情况要重试？每次重试等多久？装饰器能很好地把这两个维度塞进参数里。
先写一个最简单的重试装饰器
假设咱们只重试三次，遇到任何异常都重试，每次固定间隔1秒。装饰器长这样：
import time
import functools
def retry(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(3):
try:
return func(*args, **kwargs)
except Exception:
if attempt == 2:
raise
time.sleep(1)
return wrapper这个版本太死板，但架子搭好了。用的时候往函数头上一放：
@retry
def fetch_data():
return requests.get("https://api.example.com/data", timeout=3)可是实际项目中不同接口超时时间不一样，有的经受不住连续重试，有的要能重试5次。咱们得让参数变成可配置的。
加上参数：次数、间隔时间、捕获哪些异常
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempt = 0
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
attempt += 1
if attempt >= max_attempts:
raise
print(f"执行失败: {e}，{delay}秒后重试")
time.sleep(delay)
return wrapper
return decorator使用方式变成了：
@retry(max_attempts=5, delay=0.5, exceptions=(requests.exceptions.Timeout, requests.exceptions.ConnectionError))
def fetch_user_info(user_id):
res = requests.get(f"https://api.example.com/users/{user_id}", timeout=2)
return res.json()这样灵活多了。但你发现没有，这个delay还是固定的。如果下游服务连续挂了10秒，你固定等1秒重试，大概率每次都失败。所以指数退避就派上用场了。
真正的指数退避：越等越久，还得加抖动
指数退避的逻辑很简单：第一次等1秒，第二次等2秒，第三次等4秒，第四次等8秒……每次间隔指数增长。同时为了避免多个客户端同时重试造成“惊群效应”，一般会加一个随机抖动。
import random
def retry_with_backoff(max_attempts=5, base_delay=1, max_delay=32, exceptions=(Exception,)):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempt = 0
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
attempt += 1
if attempt >= max_attempts:
raise
# 指数退避：delay = base_delay * 2^(attempt-1)
delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
# 加抖动，防止所有请求一起重试
delay = delay + random.uniform(0, 0.1 * delay)
print(f"第{attempt}次失败: {e}，等待{delay:.2f}秒后重试")
time.sleep(delay)
return wrapper
return decorator这个版本已经能在生产环境用了。不过我发现还有个问题：如果函数是异步的（比如用httpx.AsyncClient或者aiohttp），那time.sleep会阻塞整个事件循环。必须分情况处理。
同步异步通吃的重试装饰器
要同时支持普通函数和异步函数，可以利用inspect.iscoroutinefunction来判断，然后分别返回wrapper或者async wrapper。话不多说，直接上代码：
import asyncio
import functools
import inspect
import random
import time
def retry(max_attempts=5, base_delay=1, max_delay=32, exceptions=(Exception,)):
def decorator(func):
# 如果是异步函数
if inspect.iscoroutinefunction(func):
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
attempt = 0
while True:
try:
return await func(*args, **kwargs)
except exceptions as e:
attempt += 1
if attempt >= max_attempts:
raise
delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
delay += random.uniform(0, 0.1 * delay)
print(f"异步任务第{attempt}次失败: {e}，等待{delay:.2f}秒")
await asyncio.sleep(delay)
return async_wrapper
else:
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
attempt = 0
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
attempt += 1
if attempt >= max_attempts:
raise
delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
delay += random.uniform(0, 0.1 * delay)
print(f"第{attempt}次失败: {e}，等待{delay:.2f}秒")
time.sleep(delay)
return sync_wrapper
return decorator这个装饰器拿到手就能用。同步函数就直接用time.sleep，异步函数就自动用asyncio.sleep，不会卡住事件循环。
实战案例：给爬虫接口加上优雅重试
我最近在抓某电商平台的价格，对方接口不太稳定，时不时返回一个限流错误。以前我写得贼丑：
def get_price(sku_id):
url = f"https://api.example.com/price?sku_id={sku_id}"
for _ in range(4):
try:
rsp = requests.get(url, timeout=3)
if rsp.status_code == 200:
return rsp.json()["price"]
except Exception:
time.sleep(2)
return None后来用咱刚写的装饰器重构：
@retry(max_attempts=4, base_delay=0.5, max_delay=8, exceptions=(requests.exceptions.Timeout, requests.exceptions.ConnectionError))
def get_price(sku_id):
url = f"https://api.example.com/price?sku_id={sku_id}"
rsp = requests.get(url, timeout=3)
rsp.raise_for_status()
return rsp.json()["price"]重要的几点改动：第一个是只捕获网络相关的异常，业务上返回的status_code错误让它在调用方去处理，别瞎重试。第二个是使用指数退避，第一次失败等0.5秒，第二次等1秒，第三次等2秒，这样既不会给对方服务器造成持续压力，又能快速响应瞬时抖动。
后来我发现有的接口是异步框架里的，于是直接把函数改成async def，装饰器自判断逻辑一跑，完美兼容，连测试都没改。
几个容易踩的坑，必须提醒你
1. 别捕获所有异常
像KeyboardInterrupt、SystemExit这些，继承自BaseException，不会被Exception捕获，所以默认配置下不会误伤。但你如果脑子里一热，写了个except Exception在重试装饰器里，那你可能会把代码里的Bug给吞掉重试，一直重试到死循环。一定要限制捕获异常范围。
2. 重试次数太多或延迟太长
如果你设了max_attempts=10，退避base_delay=2，最大延迟32秒，那么最坏情况要等2+4+8+16+32+32+32+32+32秒，约190秒。调用方要是没设超时，那就直接卡那儿了。合理设置，或者提供一个max_delay上限，并按需调低最大重试次数。
3. 别忘了打印日志
上面代码用了print，实际项目里建议换成logging模块。重试期间打点信息一定要有，不然别人看日志只会看到一堆等待，不知道在干嘛。
4. 幂等性
重试的前提是函数操作是幂等的。如果你重试的是一个扣款接口，那失败后强制重试可能造成重复扣款。所以装饰器一般只用在读操作或者天然支持幂等的写操作上。
进一步扩展：支持异常回调
有时候你希望每次重试后告诉某个监控平台。可以在装饰器里加上一个on_error参数，让调用方传入回调函数。
def retry(..., on_error=None):
...
except exceptions as e:
if on_error:
on_error(attempt, e, delay)
...比如这样用：
def notify(attempt, error, delay):
print(f"[监控] 第{attempt}次调用失败 {error}，{delay}秒后重试")
@retry(base_delay=1, on_error=notify)
def fetch_data():
...这个功能在报警场景里很有用，但也不是什么复杂玩意，看个人需求加。
结尾
装饰器这玩意写起来挺上头，把杂乱的重试逻辑抽象成一个注解，业务代码里就干干净净了。你现在要是再遇到那些动辄三四层循环嵌套的重试代码，可以试着拿这个装饰器去重构。别光顾着背代码，我建议你在自己项目里先跑一遍，然后调调参数，感受一下指数退避和固定间隔之间的差别。
反正我后来把项目里所有接口调用都换成了这个装饰器，再没出现过因为接口抖动导致整条链路超时的情况。你也试试吧。
