怎样实现请求重试脚本

本文目录导读：

1. 基础重试函数
2. 带退避策略的重试
3. 针对HTTP状态码的重试
4. 异步重试（使用asyncio）
5. 使用第三方库（tenacity）
6. 完整的重试脚本示例
7. 选择建议

我来介绍几种实现请求重试脚本的方法,从简单到复杂：

基础重试函数

import time
import requests
from requests.exceptions import RequestException
def request_with_retry(url, max_retries=3, delay=1):
    """
    基础请求重试函数
    """
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            return response
        except RequestException as e:
            print(f"第 {attempt + 1} 次请求失败: {e}")
            if attempt < max_retries - 1:
                time.sleep(delay * (attempt + 1))  # 递增延迟
            else:
                raise Exception("所有重试都失败")
# 使用示例
try:
    result = request_with_retry("https://api.example.com/data")
    print(result.json())
except Exception as e:
    print(f"最终失败: {e}")

带退避策略的重试

import time
import random
from functools import wraps
def retry_with_backoff(
    max_retries=3,
    base_delay=1,
    max_delay=60,
    exponential=False,
    jitter=False
):
    """
    带退避策略的重试装饰器
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    # 计算延迟时间
                    if exponential:
                        delay = min(base_delay * (2 ** attempt), max_delay)
                    else:
                        delay = base_delay * (attempt + 1)
                    # 添加抖动
                    if jitter:
                        delay = delay + random.uniform(0, delay * 0.1)
                    print(f"重试 {attempt + 1}/{max_retries}, 等待 {delay:.2f}秒")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator
# 使用示例
@retry_with_backoff(max_retries=5, exponential=True, jitter=True)
def fetch_data(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()
# 调用
data = fetch_data("https://api.example.com/data")

针对HTTP状态码的重试

import time
import requests
class HTTPRetryHandler:
    """
    基于HTTP状态码的请求重试
    """
    RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
    def __init__(self, max_retries=3, delay=1):
        self.max_retries = max_retries
        self.delay = delay
    def request(self, method, url, **kwargs):
        """
        执行带重试的HTTP请求
        """
        for attempt in range(self.max_retries):
            try:
                response = requests.request(method, url, **kwargs)
                # 检查是否需要重试
                if response.status_code in self.RETRYABLE_STATUS_CODES:
                    if attempt < self.max_retries - 1:
                        retry_after = self._get_retry_after(response)
                        wait_time = retry_after or (self.delay * (attempt + 1))
                        print(f"状态码 {response.status_code}, 等待 {wait_time}秒后重试")
                        time.sleep(wait_time)
                        continue
                response.raise_for_status()
                return response
            except requests.exceptions.ConnectionError:
                if attempt < self.max_retries - 1:
                    print(f"连接错误, {self.delay * (attempt + 1)}秒后重试")
                    time.sleep(self.delay * (attempt + 1))
                else:
                    raise
            except requests.exceptions.Timeout:
                if attempt < self.max_retries - 1:
                    print(f"超时错误, {self.delay * (attempt + 1)}秒后重试")
                    time.sleep(self.delay * (attempt + 1))
                else:
                    raise
        raise Exception(f"所有重试都失败, 最后状态码: {response.status_code}")
    def _get_retry_after(self, response):
        """获取Retry-After头信息"""
        retry_after = response.headers.get('Retry-After')
        if retry_after:
            try:
                return int(retry_after)
            except ValueError:
                pass
        return None
# 使用示例
retry_handler = HTTPRetryHandler(max_retries=5, delay=2)
response = retry_handler.request('GET', 'https://api.example.com/data')

异步重试（使用asyncio）

import asyncio
import aiohttp
import time
async def async_retry_request(url, max_retries=3, delay=1):
    """
    异步请求重试
    """
    async with aiohttp.ClientSession() as session:
        for attempt in range(max_retries):
            try:
                async with session.get(url, timeout=10) as response:
                    if response.status == 200:
                        return await response.json()
                    elif response.status in [429, 500, 502, 503, 504]:
                        if attempt < max_retries - 1:
                            delay_time = delay * (attempt + 1)
                            print(f"状态码 {response.status}, 等待 {delay_time}秒")
                            await asyncio.sleep(delay_time)
                            continue
                    else:
                        response.raise_for_status()
            except asyncio.TimeoutError:
                if attempt < max_retries - 1:
                    await asyncio.sleep(delay * (attempt + 1))
                else:
                    raise
            except Exception as e:
                if attempt < max_retries - 1:
                    print(f"错误: {e}, 重试中...")
                    await asyncio.sleep(delay * (attempt + 1))
                else:
                    raise
        raise Exception("所有重试都失败")
# 并行执行多个重试请求
async def main():
    urls = [
        "https://api1.example.com/data",
        "https://api2.example.com/data",
        "https://api3.example.com/data"
    ]
    tasks = [async_retry_request(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            print(f"{url} 失败: {result}")
        else:
            print(f"{url} 成功: {result}")
# 运行
# asyncio.run(main())

使用第三方库（tenacity）

from tenacity import retry, stop_after_attempt, wait_exponential
import requests
# 最简单的使用方式
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def make_request(url):
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()
# 更高级的配置
@retry(
    stop=stop_after_attempt(5),  # 最多重试5次
    wait=wait_exponential(multiplier=2, min=1, max=60),  # 指数退避
    retry_error_callback=lambda retry_state: print(f"重试 {retry_state.attempt_number}")  # 回调
)
def fetch_data_with_retry(url):
    response = requests.get(url, timeout=10)
    # 自定义重试条件
    if response.status_code in [429, 503]:
        raise Exception(f"Server overloaded: {response.status_code}")
    response.raise_for_status()
    return response.json()
# 使用
try:
    data = make_request("https://api.example.com/data")
except Exception as e:
    print(f"所有重试都失败: {e}")

完整的重试脚本示例

#!/usr/bin/env python3
"""
完整的请求重试脚本
支持：状态码重试、连接错误、超时、指数退避、日志记录
"""
import time
import logging
import requests
from requests.exceptions import (
    RequestException,
    ConnectionError,
    Timeout,
    HTTPError
)
from typing import Optional, Dict, Any
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class SmartRetryClient:
    """
    智能重试HTTP客户端
    """
    RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504, 509}
    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        backoff_type: str = 'exponential',
        timeout: int = 30
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.backoff_type = backoff_type
        self.timeout = timeout
        self.session = requests.Session()
    def get_delay(self, attempt: int) -> float:
        """计算重试延迟"""
        if self.backoff_type == 'exponential':
            delay = min(self.base_delay * (2 ** attempt), self.max_delay)
        elif self.backoff_type == 'linear':
            delay = min(self.base_delay * (attempt + 1), self.max_delay)
        else:  # constant
            delay = self.base_delay
        return delay
    def request(
        self,
        method: str,
        url: str,
        **kwargs
    ) -> Optional[requests.Response]:
        """
        带重试的请求方法
        """
        last_exception = None
        for attempt in range(self.max_retries):
            try:
                logger.info(f"请求 {attempt + 1}/{self.max_retries}: {method} {url}")
                response = self.session.request(
                    method,
                    url,
                    timeout=self.timeout,
                    **kwargs
                )
                # 检查状态码是否可重试
                if response.status_code in self.RETRYABLE_STATUS_CODES:
                    if attempt < self.max_retries - 1:
                        retry_after = response.headers.get('Retry-After')
                        wait_time = float(retry_after) if retry_after else self.get_delay(attempt)
                        logger.warning(
                            f"状态码 {response.status_code}, "
                            f"等待 {wait_time:.2f}秒后重试"
                        )
                        time.sleep(wait_time)
                        continue
                    else:
                        response.raise_for_status()
                # 成功响应
                logger.info(f"请求成功: {response.status_code}")
                return response
            except (ConnectionError, Timeout) as e:
                last_exception = e
                if attempt < self.max_retries - 1:
                    delay = self.get_delay(attempt)
                    logger.warning(
                        f"连接错误: {e}, {delay:.2f}秒后重试"
                    )
                    time.sleep(delay)
                else:
                    logger.error(f"所有重试都失败: {e}")
                    raise
            except HTTPError as e:
                if attempt < self.max_retries - 1 and e.response.status_code in self.RETRYABLE_STATUS_CODES:
                    delay = self.get_delay(attempt)
                    logger.warning(
                        f"HTTP错误 {e.response.status_code}, "
                        f"{delay:.2f}秒后重试"
                    )
                    time.sleep(delay)
                else:
                    raise
            except RequestException as e:
                last_exception = e
                if attempt < self.max_retries - 1:
                    delay = self.get_delay(attempt)
                    logger.warning(f"请求异常: {e}, {delay:.2f}秒后重试")
                    time.sleep(delay)
                else:
                    raise
        raise last_exception if last_exception else Exception("请求失败")
    def get(self, url: str, **kwargs) -> Optional[requests.Response]:
        return self.request('GET', url, **kwargs)
    def post(self, url: str, **kwargs) -> Optional[requests.Response]:
        return self.request('POST', url, **kwargs)
# 使用示例
def main():
    # 创建客户端
    client = SmartRetryClient(
        max_retries=5,
        base_delay=1,
        backoff_type='exponential',
        timeout=10
    )
    # 示例请求
    try:
        response = client.get("https://api.github.com")
        print("状态码:", response.status_code)
        print("数据:", response.json()[:100])
    except Exception as e:
        print(f"最终失败: {e}")
    # 测试重试（使用不会响应的URL）
    try:
        response = client.get("https://api.example.com/nonexistent")
    except Exception as e:
        print(f"预期失败: {e}")
if __name__ == "__main__":
    main()

选择建议

1. 简单场景：使用基础的retry_with_backoff函数
2. 需要精细控制：使用SmartRetryClient类
3. 异步场景：使用asyncio版本
4. 不想自己写：使用tenacity库
5. 生产环境：建议使用完整的SmartRetryClient实现，包含日志、超时、退避策略等

选择哪种方式取决于你的具体需求和使用的技术栈。