Python Requests实现API请求重试与超时配置全解析

更新时间：2025年11月28日 08:21:22 作者：站大爷IP

API请求的稳定性直接决定系统可靠性,本文通过真实案例拆解,Python如何用Requests库实现防抖动+抗异常的健壮请求方案,需要的可以了解下

在电商物流追踪、金融数据监控等场景中，API请求的稳定性直接决定系统可靠性。当顺丰API因网络抖动返回503错误，或因跨地域调用出现10秒延迟时，如何确保程序不崩溃且数据不丢失？本文通过真实案例拆解，用Requests库实现"防抖动+抗异常"的健壮请求方案。

一、血泪教训：那些年踩过的API坑

某跨境电商系统在"黑色星期五"大促期间突发故障：调用顺丰国际件接口时，30%的请求因超时失败，导致2000+包裹状态同步延迟。事后分析发现三大元凶：

固定超时陷阱：设置timeout=5导致所有跨洋请求必然超时（实际平均响应时间8秒）
暴力重试雪崩：简单for循环重试5次，瞬间产生10倍请求量击垮顺丰网关
代理池污染：使用失效代理IP发起请求，触发顺丰反爬机制封禁整个IP段

这些场景揭示核心问题：API请求需要"有智慧的等待"和"有策略的坚持" 。

二、超时配置：给请求装上"安全阀"

1. 连接超时 vs 读取超时

import requests

try:
    # 连接超时3秒（TCP握手阶段）
    # 读取超时10秒（服务器处理阶段）
    response = requests.get(
        'https://api.sf-express.com/track',
        params={'trackingNumber': 'SF123456789'},
        timeout=(3, 10)  # 元组形式分别设置
    )
    print(response.json())
except requests.exceptions.ConnectTimeout:
    print("连接服务器失败，请检查网络")
except requests.exceptions.ReadTimeout:
    print("服务器处理超时，请稍后重试")

关键决策点：

国内API调用：timeout=(2, 5)（连接2秒，读取5秒）
跨境API调用：timeout=(5, 15)（考虑国际链路延迟）
文件上传场景：需增加write_timeout参数（需httpx等库支持）

2. 动态超时策略

某物流监控系统采用分级超时机制：

def get_dynamic_timeout(retry_count):
    base_timeout = 3  # 基础超时
    if retry_count > 0:
        return min(base_timeout * (2 ** retry_count), 30)  # 指数退避，最大30秒
    return base_timeout

# 使用示例
for i in range(3):
    try:
        timeout = get_dynamic_timeout(i)
        response = requests.get(url, timeout=timeout)
        break
    except Exception as e:
        print(f"第{i+1}次尝试失败，超时时间调整为{timeout}秒")

三、重试机制：让请求学会"坚持"

1. 指数退避重试（推荐方案）

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session(retries=3, backoff_factor=1, status_forcelist=(500, 502, 503, 504)):
    session = requests.Session()
    retry = Retry(
        total=retries,
        read=True,  # 允许读取超时重试
        connect=True,  # 允许连接超时重试
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        allowed_methods=["GET", "POST"]  # 支持POST请求重试
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

# 使用示例
session = create_retry_session()
response = session.get('https://api.sf-express.com/track', params={'trackingNumber': 'SF123456789'})

参数深解：

backoff_factor=1：第1次重试等待1秒，第2次2秒，第3次4秒
status_forcelist：仅对5xx服务器错误和429限流错误重试
allowed_methods：默认不重试POST请求，需显式声明

2. 熔断机制实现（避免雪崩）

from collections import deque
import time

class CircuitBreaker:
    def __init__(self, max_failures=3, reset_timeout=60):
        self.failures = deque(maxlen=max_failures)
        self.reset_timeout = reset_timeout

    def is_open(self):
        if len(self.failures) < self.failures.maxlen:
            return False
        # 如果最近max_failures次请求都失败，且最后一次失败在reset_timeout秒内
        return (time.time() - self.failures[-1]) < self.reset_timeout

    def record_failure(self):
        self.failures.append(time.time())

    def record_success(self):
        self.failures.clear()

# 结合重试使用
breaker = CircuitBreaker(max_failures=3, reset_timeout=30)

def safe_request():
    if breaker.is_open():
        raise Exception("Service unavailable, circuit breaker open")
    
    try:
        response = create_retry_session().get(url)
        if response.status_code == 200:
            breaker.record_success()
            return response
        else:
            breaker.record_failure()
            raise Exception("API request failed")
    except Exception as e:
        breaker.record_failure()
        raise e

四、代理配置：突破封禁的"隐身术"

1. 代理池实战方案

import random
from requests.adapters import HTTPAdapter

class ProxyPool:
    def __init__(self):
        self.proxies = [
            {"http": "http://1.1.1.1:8080", "https": "http://1.1.1.1:8080"},
            {"http": "http://2.2.2.2:8080", "https": "http://2.2.2.2:8080"},
            # 更多代理...
        ]
        self.failed_proxies = set()

    def get_proxy(self):
        available_proxies = [p for p in self.proxies if p not in self.failed_proxies]
        if not available_proxies:
            raise Exception("No available proxies")
        return random.choice(available_proxies)

    def mark_failed(self, proxy):
        self.failed_proxies.add(proxy)

# 使用示例
proxy_pool = ProxyPool()
session = requests.Session()

for _ in range(3):  # 尝试3个不同代理
    try:
        proxy = proxy_pool.get_proxy()
        response = session.get(
            'https://api.sf-express.com/track',
            proxies=proxy,
            timeout=(3, 10)
        )
        if response.status_code == 200:
            print("Success with proxy:", proxy)
            break
        else:
            proxy_pool.mark_failed(proxy)
    except Exception:
        proxy_pool.mark_failed(proxy)
else:
    print("All proxies failed")

2. Tor代理配置（高匿名场景）

import requests

def make_tor_request(url):
    proxies = {
        'http': 'socks5h://127.0.0.1:9050',
        'https': 'socks5h://127.0.0.1:9050'
    }
    try:
        response = requests.get(url, proxies=proxies, timeout=(5, 15))
        print("Tor出口IP:", response.json()['origin'])
        return response
    except Exception as e:
        print("Tor请求失败:", e)

# 需提前安装Tor服务并启动
# sudo apt install tor  # Ubuntu系统
# sudo service tor start

五、完整实战案例

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
from datetime import datetime

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class SFExpressTracker:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = self._create_session()

    def _create_session(self):
        """创建带重试和超时的会话"""
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session = requests.Session()
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session

    def _generate_sign(self, params):
        """生成API签名（简化版）"""
        import hashlib
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        raw_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
        return hashlib.md5(raw_str.encode()).hexdigest().upper()

    def query_track(self, tracking_number):
        """查询物流轨迹"""
        url = "https://bsp-ois.sf-express.com/bsp-ois/express/service/queryTrack"
        params = {
            "appKey": self.app_key,
            "trackNumber": tracking_number,
            "timestamp": str(int(datetime.now().timestamp()))
        }
        params["sign"] = self._generate_sign(params)

        try:
            response = self.session.get(
                url,
                params=params,
                timeout=(3, 10)  # 连接3秒，读取10秒
            )
            response.raise_for_status()
            data = response.json()
            
            if data.get('success'):
                return data['data']['tracks']
            else:
                logging.warning(f"API返回错误: {data.get('errorMsg', '未知错误')}")
                return None
                
        except requests.exceptions.RequestException as e:
            logging.error(f"请求失败: {str(e)}")
            return None

# 使用示例
if __name__ == "__main__":
    tracker = SFExpressTracker(app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET")
    tracks = tracker.query_track("SF123456789")
    if tracks:
        for step in tracks:
            print(f"{step['acceptTime']} {step['acceptAddress']} - {step['remark']}")

六、常见问题Q&A

Q1：被网站封IP怎么办？

A：立即启用备用代理池，建议使用住宅代理（如站大爷IP代理），配合每请求更换IP策略。对于大规模爬取，可采用Tor网络或IP轮换中间件。

Q2：如何选择重试次数？

A：遵循"3次黄金法则"：

首次请求
指数退避重试2次（总计3次）
超过3次仍失败应触发熔断或人工干预

Q3：代理请求速度慢怎么解决？

A：

测试代理延迟：curl --socks5 127.0.0.1:9050 https://httpbin.org/ip
使用代理评分机制，淘汰高延迟代理
对代理请求添加User-Agent和常规请求头

Q4：如何记录重试日志？

A：扩展Retry类实现自定义日志：

from urllib3.util.retry import Retry
import logging

class LoggingRetry(Retry):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger(__name__)

    def new(self, **kw):
        self.logger.debug(f"Creating new retry adapter with params: {kw}")
        return super().new(**kw)

    def increment(self, method, url, response=None, error=None, **kwargs):
        self.logger.warning(f"Retry attempt {self.total - self._remaining + 1} for {method} {url}")
        return super().increment(method, url, response, error, **kwargs)

Q5：POST请求重试需要注意什么？

A：

确保请求是幂等的（如使用唯一请求ID）
在重试前检查响应是否已部分处理
考虑使用idempotency-key请求头（如Stripe API要求）

通过合理组合超时配置、智能重试和代理策略，可构建出应对各种异常场景的健壮API请求系统。实际开发中建议结合Prometheus监控重试率、失败率等指标，持续优化请求策略。

以上就是Python Requests实现API请求重试与超时配置全解析的详细内容，更多关于Python Requests请求重试与超时配置的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

python基础之字典
这篇文章主要介绍了python的字典,实例分析了Python中返回一个返回值与多个返回值的方法,需要的朋友可以参考下
2021-10-10
python3的url编码和解码,自定义gbk、utf-8的例子
今天小编就为大家分享一篇python3的url编码和解码,自定义gbk、utf-8的例子，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2019-08-08
python映射列表实例分析
这篇文章主要介绍了python映射列表,实例分析了python映射列表遍历计算其中每一个元素的使用技巧,需要的朋友可以参考下
2015-01-01
PyQt5类型判定+对象删除操作
这篇文章主要介绍了PyQt5类型判定+对象删除操作,本文通过实例代码给大家介绍的非常详细,感兴趣的朋友跟随小编一起看看吧
2024-06-06
使用npy转image图像并保存的实例
这篇文章主要介绍了使用npy转image图像并保存的实例，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2020-07-07
python中JSON数据格式的详细使用教程
这篇文章主要给大家介绍了关于python中JSON数据格式的详细使用,JSON是一种用于存储和交换数据的语法,JSON是文本,使用JavaScript对象表示法编写,文中通过代码介绍的非常详细,需要的朋友可以参考下
2024-02-02
基于Python实现植物大战僵尸游戏的示例代码
植物大战僵尸是一款经典的塔防类游戏,玩家通过种植各种植物来抵御僵尸的攻击,本文将详细介绍如何使用Python和Pygame库来实现一个简单的植物大战僵尸游戏,文中通过代码示例讲解的非常详细,感兴趣的小伙伴跟着小编一起来看看吧
2024-10-10
Python实现音频提取的示例详解
在日常生活中,有好听的翻唱视频或音乐视频可以将其音频分离保存到网易云或QQ音乐中随时听,然而大部分的音频分离软件需要下载和安装,所以本文就来分享一种提取音频的简便方法吧
2023-09-09
python实现邻接表转邻接矩阵
这篇文章主要介绍了python实现邻接表转邻接矩阵，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-12-12
Python中的 ansible 动态Inventory 脚本
这篇文章主要介绍了Python中的 ansible 动态Inventory 脚本,本章节通过实例代码从mysql数据作为数据源生成动态ansible主机为入口介绍的非常详细，感兴趣的朋友跟随小编一起看看吧
2020-01-01