Python请求重试案例怎么编写？

wen python案例 2026-06-08 10:07:12 1

高效编写Python请求重试案例：从入门到精通的完整指南

目录导读

为什么需要请求重试机制？
Python请求重试的常见应用场景
核心实践：四种主流重试案例编写方法
高级技巧：自定义重试逻辑与策略
常见问题与解答
总结与最佳实践

为什么需要请求重试机制？

在网络编程中，HTTP请求失败是家常便饭，无论是网络抖动、服务器临时过载，还是DNS解析异常，都可能导致请求中断。编写可靠的请求重试案例，本质上是为你的应用穿上“防弹衣”——当第一次请求失败时，系统能自动进行后续重试,避免因单点故障导致整个业务流程崩溃。

问题： 不写重试机制会有什么后果？
回答： 会导致应用脆弱性陡增，比如爬虫在抓取数据时，因一次网络超时直接退出；API调用因服务端偶发错误导致事务中断，缺乏重试的代码，在生成环境中故障率会增加30%以上。

Python请求重试的常见应用场景

网络不稳定环境：移动端采集、跨境API调用（如跨境支付接口）
高并发下的微服务：下游服务偶发503或429错误
爬虫与数据采集：防反爬策略中，因IP限制或验证码导致的请求失败
文件上传/下载：大文件传输中断后自动续传

核心实践：四种主流重试案例编写方法

使用`requests`结合`time.sleep`手动重试（基础版）

import requests
import time
def request_with_retry(url, max_retries=3, delay=2):
    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                return response.json()
            else:
                print(f"状态码异常: {response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"第{attempt}次请求失败: {e}")
            if attempt < max_retries:
                time.sleep(delay)
    raise Exception("超过最大重试次数，请求失败")

适用场景：简单脚本、单线程任务
缺点：代码重复，不支持指数退避

`urllib3.Retry`+`requests.Session`（中级版）

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import requests
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,  # 等待时间 = backoff_factor * (2 ** (重试次数-1))
    status_forcelist=[502, 503, 504],  # 仅对这些状态码重试
    allowed_methods=["GET", "POST"]    # 允许重试的方法
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)
response = session.get("https://api.example.com/data")

核心优势：自动实现指数退避，无需手动sleep；可单独指定状态码策略
注意：需安装urllib3 >= 1.26

`tenacity`库（高级版）

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=2, min=1, max=10),
    retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout)),
    before_sleep=lambda retry_state: print(f"重试{retry_state.attempt_number}次后等待{retry_state.outcome.exception()}")
)
def fetch_data(url):
    response = requests.get(url, timeout=3)
    response.raise_for_status()
    return response.json()
data = fetch_data("https://api.example.com/resource")

独特价值：

支持重试条件自定义（按异常类型、返回结果、甚至自定义函数）
内置重试统计回调（如记录每次重试原因）
可结合异步重试（需tenacity.asyncio）

`httpx`异步重试（高性能版）

import httpx
from tenacity import retry, stop_after_attempt
client = httpx.AsyncClient(timeout=10)
@retry(stop=stop_after_attempt(3))
async def async_fetch(url):
    async with client as c:
        resp = await c.get(url)
        resp.raise_for_status()
        return resp.json()
# 使用 asyncio.run(async_fetch("https://api.example.com"))

适用场景：高并发爬虫、实时数据流处理
性能提升：比同步重试效率提升5~10倍

高级技巧：自定义重试逻辑与策略

1 根据响应状态码动态调整重试行为

from urllib3.util.retry import Retry
class CustomRetry(Retry):
    def is_retry(self, method, status_code, has_retry_after=False):
        if status_code == 429:  # 限流时增加延迟
            return True
        return super().is_retry(method, status_code, has_retry_after)

2 实现“熔断+渐进式重试”

import random
def smart_retry(url, max_attempts=5, base_delay=1):
    for i in range(max_attempts):
        try:
            resp = requests.get(url)
            if resp.status_code < 500:
                return resp
        except:
            delay = base_delay * (2 ** i) + random.uniform(0, 1)
            print(f"等待{delay:.2f}秒")
            time.sleep(delay)
    return None

3 集成日志与监控

在重试过程中记录关键指标：

import logging
logger = logging.getLogger(__name__)
@retry(stop=stop_after_attempt(3))
def monitored_request(url):
    try:
        response = requests.get(url)
        logger.info(f"成功请求 {url} 状态码: {response.status_code}")
        return response
    except Exception as e:
        logger.warning(f"请求失败: {e}")
        raise

常见问题与解答

Q1：重试会导致雪崩效应吗？
A：会！当大量请求同时失败并触发重试，可能压垮上游服务，解决方案：

加入Jitter（随机抖动）避免重试请求集中在同一时间戳
使用熔断机制，当连续失败达到阈值时暂停重试一段时间

Q2：哪些异常不应该重试？
A：不应该重试的异常包括：

requests.exceptions.InvalidURL（URL格式错误）
requests.exceptions.TooManyRedirects（重定向循环）
HTTP状态码400、401、403、404（客户端错误,重试无效）

Q3：如何测试重试逻辑是否正确？
A：推荐使用pytest配合responses库模拟失败/成功响应：

import responses
@responses.activate
def test_retry_success():
    # 前两次返回503，第三次返回200
    responses.add(responses.GET, "https://api.example.com", status=503, body="Server Error")
    responses.add(responses.GET, "https://api.example.com", status=503, body="Server Error")
    responses.add(responses.GET, "https://api.example.com", status=200, body="OK")
    result = fetch_with_retry("https://api.example.com")
    assert result == "OK"

Q4：异步重试中如何保证线程安全？
A：使用asyncio.Lock同步共享资源，或者为每个协程创建独立的httpx.AsyncClient实例。