https://github.com/jd/tenacity 是python生态里一个很棒的重试库，基本用法大家可去查看他们的官网
背景
要写个爬虫，要抓的数据是分页的，每次请求都带一个page参数，如果请求失败，就重试，重试的时候，page参数要保持，不能丢失
默认情况 tenacity 每次重试，都会用 函数首次传入的参数去重试， 如果我请求数据到23页，这时候报错， 它触发重试机制的话， 就会从第1页开始重试， 这没有意义
错误示范
import logging
import random
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
# 设置日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def log_retry_attempt(retry_state):
"""记录重试信息的回调函数"""
# 注意，使用重试的函数，上游调用它的时候传参必须k=v格式传参，不能用位置传参，不然kwargs里什么都拿不到
page = retry_state.kwargs.get('page')
exception = retry_state.outcome.exception()
logger.warning(
f"正在重试 {retry_state.fn.__name__} "
f"(第 {retry_state.attempt_number} 次尝试) "
f"针对 page:{page}。"
f"将在{retry_state.next_action.sleep} 秒后重试... "
f"异常: {str(exception)}"
)
# 这里重试的时候，它会用上游调用它的时候，传递的参数来重试，上游我们传递的参数page=1，所以不论尝试多少次，page都是1，这样死循环是没有意义的
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
before_sleep=log_retry_attempt,
reraise=True
)
def fetch_data(page):
all_data = []
while True:
try:
print(f"正在请求第{page}页数据")
# 模拟一个API请求，假设请求失败
response = requests.get(f'https://api.restful-api.dev/objects/{page}')
response.raise_for_status()
data = response.json()
# 随机抛出错误
if page > 2: # 第二页以后才随机出错
if random.random() < 0.3: # 30% 概率抛出错误
raise ValueError("模拟的随机错误")
all_data.extend(data)
except requests.exceptions.RequestException as e:
# 如果请求失败，抛出异常，重试
raise e
page = page + 1
pass
if __name__ == "__main__":
res = fetch_data(page=1)
print(res)
错误示范截图
正确示范
import logging
import random
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
# 设置日志配置
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def log_retry_attempt(retry_state):
"""记录重试信息的回调函数"""
# 注意，使用重试的函数，上游调用它的时候传参必须k=v格式传参，不能用位置传参，不然kwargs里什么都拿不到
page = retry_state.kwargs.get('page')
exception = retry_state.outcome.exception()
logger.warning(
f"正在重试 {retry_state.fn.__name__} "
f"(第 {retry_state.attempt_number} 次尝试) "
f"针对 page:{page}。"
f"将在{retry_state.next_action.sleep} 秒后重试... "
f"异常: {str(exception)}"
)
# 这里重试的时候，也是用上游调用它的时候，传递的参数来重试，上游我们传递的参数page在fetch_all_data里，每一次都不一样，某次报错，就会用错误的那一轮的page重试，这样就刚好符合我们的需求
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
before_sleep=log_retry_attempt,
reraise=True
)
def fetch_data(page):
print(f"正在请求第{page}页数据")
# 随机抛出错误
if page > 2: # 第二页以后才随机出错
if random.random() < 0.3: # 30% 概率抛出错误
raise ValueError("模拟的随机错误")
response = requests.get(f'https://api.restful-api.dev/objects/{page}')
response.raise_for_status()
data = response.json()
return data
def fetch_all_data(page):
all_data = []
while True:
try:
data = fetch_data(page=page) # 每次调用，传递的page都不同，如果请求失败，就会用这个page重试，成功后，我们就会用新的page调用fetch_data， 完美符合需求
all_data.extend(data)
page += 1
except Exception as e:
print(e)
break
return all_data
if __name__ == "__main__":
res = fetch_all_data(page=1)
print(res)
正确示范截图
总结
其实就是 获取数据 专门独立为一个函数，里面不需要什么容错语句，就是单纯粗暴的获取数据并返回，然后给这个函数头上挂重试的装饰器，这样当这个函数被上游调用出错的时候，就会用参数page重试 （比如 正确示范里的 fetch_data 方法）
而上游的方法（比如 正确示范里的 fetch_all_data 方法 ），则必须 用循环的方式， 处理好 page， 每次用不同的参数 page 去调用有重试机制的函数