Python接口重试案例如何失败重试请求

wen python案例 2026-06-30 188

本文目录导读：

使用 requests + urllib3 的重试机制
使用 tenacity 库（更灵活的重试方案）
手动实现重试机制（自定义控制）
完整的重试案例（包含所有常见场景）
重试策略建议

在Python中实现接口请求的失败重试,主要有以下几种常见方案：

使用 `requests` + `urllib3` 的重试机制

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_retry_session(retries=3, backoff_factor=0.5):
    """
    创建带重试机制的请求会话
    retries: 重试次数
    backoff_factor: 退避因子（每次重试等待时间 = backoff_factor * (2 ** (重试次数-1))）
    """
    session = requests.Session()
    retry_strategy = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=[500, 502, 503, 504],  # 需要重试的HTTP状态码
        allowed_methods=["GET", "POST", "PUT", "DELETE"],  # 允许重试的HTTP方法
        raise_on_status=True  # 超过重试次数后抛出异常
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session
# 使用示例
def fetch_with_retry(url, headers=None, timeout=10):
    session = create_retry_session(retries=3, backoff_factor=1)
    try:
        response = session.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()  # 检查请求是否成功
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败，已重试3次: {e}")
        return None
    finally:
        session.close()
# 测试
if __name__ == "__main__":
    # 模拟一个可能失败的服务
    result = fetch_with_retry("https://api.example.com/data")
    print(result)

使用 `tenacity` 库（更灵活的重试方案）

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数退避等待
    retry=retry_if_exception_type((requests.ConnectionError, 
                                   requests.Timeout,
                                   requests.HTTPError))  # 只在这些异常时重试
)
def fetch_data_with_tenacity(url):
    """使用tenacity的重试装饰器"""
    response = requests.get(url, timeout=5)
    response.raise_for_status()  # 如果HTTP状态码>=400则抛出异常
    return response.json()
# 更复杂的重试逻辑
from tenacity import before_log, after_log, before_sleep_log
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((requests.exceptions.RequestException,)),
    before=before_log(logger, logging.INFO),
    after=after_log(logger, logging.INFO),
    before_sleep=before_sleep_log(logger, logging.INFO),
    reraise=True  # 重试结束后重新抛出异常
)
def fetch_with_logging(url):
    """带日志记录的重试请求"""
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    data = response.json()
    return data
# 使用示例
if __name__ == "__main__":
    try:
        result = fetch_with_logging("https://api.example.com/data")
        print(result)
    except Exception as e:
        print(f"最终请求失败: {e}")

手动实现重试机制（自定义控制）

import requests
import time
from typing import Callable, Any, Optional
def retry_request(
    func: Callable,
    max_retries: int = 3,
    backoff_factor: float = 0.5,
    exceptions: tuple = (requests.exceptions.RequestException,),
    on_retry: Optional[Callable] = None
) -> Any:
    """
    通用的重试装饰器
    func: 要重试的函数
    max_retries: 最大重试次数
    backoff_factor: 退避因子
    exceptions: 需要重试的异常类型
    on_retry: 重试时的回调函数
    """
    for attempt in range(max_retries + 1):
        try:
            return func()
        except exceptions as e:
            if attempt == max_retries:
                raise  # 最后一次尝试失败，抛出异常
            wait_time = backoff_factor * (2 ** attempt)
            if on_retry:
                on_retry(attempt + 1, max_retries, e, wait_time)
            else:
                print(f"重试 {attempt + 1}/{max_retries}, 等待 {wait_time:.2f}秒, 错误: {e}")
            time.sleep(wait_time)
# 自定义重试回调
def custom_on_retry(attempt, max_retries, error, wait_time):
    """自定义重试时的行为"""
    print(f"[{attempt}/{max_retries}] 请求失败: {error}")
    print(f"等待 {wait_time:.2f}秒后重试...")
# 使用示例
def make_request():
    """要重试的实际请求函数"""
    response = requests.get("https://api.example.com/data", timeout=5)
    response.raise_for_status()
    return response.json()
# 执行带重试的请求
if __name__ == "__main__":
    try:
        result = retry_request(
            make_request,
            max_retries=3,
            backoff_factor=1,
            on_retry=custom_on_retry
        )
        print(f"请求成功: {result}")
    except Exception as e:
        print(f"最终请求失败: {e}")

完整的重试案例（包含所有常见场景）

import requests
import time
import logging
from functools import wraps
from typing import Union, Dict, Optional
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class RetryStrategy:
    """重试策略配置"""
    def __init__(
        self,
        max_retries: int = 3,
        backoff_factor: float = 0.5,
        max_backoff: float = 30,
        retry_on_status: list = None,
        retry_on_exceptions: tuple = None,
        allowed_methods: list = None
    ):
        self.max_retries = max_retries
        self.backoff_factor = backoff_factor
        self.max_backoff = max_backoff
        self.retry_on_status = retry_on_status or [500, 502, 503, 504]
        self.retry_on_exceptions = retry_on_exceptions or (
            requests.ConnectionError,
            requests.Timeout,
            requests.HTTPError
        )
        self.allowed_methods = allowed_methods or ["GET", "POST", "PUT", "DELETE"]
def retry_decorator(strategy: RetryStrategy = None):
    """重试装饰器"""
    if strategy is None:
        strategy = RetryStrategy()
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(strategy.max_retries + 1):
                try:
                    response = func(*args, **kwargs)
                    # 检查HTTP状态码
                    if response.status_code in strategy.retry_on_status:
                        if attempt == strategy.max_retries:
                            response.raise_for_status()
                        raise requests.HTTPError(
                            f"HTTP {response.status_code}",
                            response=response
                        )
                    return response
                except strategy.retry_on_exceptions as e:
                    last_exception = e
                    if attempt == strategy.max_retries:
                        logger.error(f"达到最大重试次数 {strategy.max_retries}, 请求失败")
                        raise
                    # 计算等待时间（指数退避）
                    wait_time = min(
                        strategy.backoff_factor * (2 ** attempt),
                        strategy.max_backoff
                    )
                    logger.warning(
                        f"请求失败 (尝试 {attempt + 1}/{strategy.max_retries}), "
                        f"等待 {wait_time:.2f}秒, 错误: {type(e).__name__}: {e}"
                    )
                    time.sleep(wait_time)
            # 不应该执行到这里，但为了安全返回
            if last_exception:
                raise last_exception
        return wrapper
    return decorator
# 使用装饰器的示例
@retry_decorator()
def fetch_data(url: str, headers: Dict = None, timeout: int = 10) -> requests.Response:
    """获取数据（带重试机制）"""
    logger.info(f"请求 URL: {url}")
    response = requests.get(url, headers=headers, timeout=timeout)
    response.raise_for_status()
    return response
# 更具体的配置
custom_strategy = RetryStrategy(
    max_retries=5,
    backoff_factor=1,
    max_backoff=60,
    retry_on_status=[429, 500, 502, 503, 504],  # 包含429限流状态码
    retry_on_exceptions=(requests.ConnectionError, requests.Timeout)
)
@retry_decorator(custom_strategy)
def fetch_with_custom_strategy(url: str) -> requests.Response:
    """使用自定义策略的请求"""
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response
# 测试用例
if __name__ == "__main__":
    # 测试正常的请求
    try:
        response = fetch_data("http://httpbin.org/get")
        print(f"成功获取数据: {response.status_code}")
    except Exception as e:
        print(f"请求失败: {e}")
    # 测试模拟失败的服务
    try:
        response = fetch_with_custom_strategy("http://httpbin.org/status/500")
        print(f"状态码: {response.status_code}")
    except Exception as e:
        print(f"请求最终失败: {e}")
    # 测试超时重试
    try:
        response = fetch_data("http://httpbin.org/delay/10", timeout=3)
    except requests.Timeout:
        print("请求超时并重试后仍然失败")

重试策略建议

何时使用哪种方案：

简单场景：使用 requests + urllib3 的重试机制
需要灵活控制：使用 tenacity 库
需要完全控制：手动实现重试逻辑

重试注意事项：

幂等性：确保重试的请求是幂等的（不会产生副作用）
超时设置：合理设置请求超时时间
退避策略：使用指数退避避免雪崩效应
日志记录：记录重试过程便于排查问题
限流保护：对429状态码也要考虑重试

最佳实践：

# 建议使用的完整配置
retry_config = {
    'max_retries': 3,          # 最多重试3次
    'backoff_factor': 0.5,     # 退避因子
    'max_backoff': 10,         # 最大等待时间
    'status_forcelist': [429, 500, 502, 503, 504],
    'allowed_methods': ['GET', 'HEAD', 'OPTIONS'],  # 只对幂等方法重试
    'raise_on_status': True    # 超过重试次数后抛出异常
}

这些案例涵盖了Python中常见的接口重试实现方式,你可以根据实际需求选择最适合的方案。

本文地址： https://www.bmjup.com/post/12478.html