You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python requests库中配置多冗余REST API主机列表?

在requests客户端实现冗余设备自动切换

完全可以在Python requests(或其底层urllib3)中实现客户端侧的冗余设备自动选路,不需要额外配置负载均衡器。核心思路是通过自定义Session扩展或包装请求逻辑,遍历目标设备列表直到找到可用节点。

方案1:自定义故障转移Session(推荐)

创建继承自requests.Session的类,内置目标设备列表,重写request方法实现故障转移逻辑。这种方式简单直接,完全基于requests原生能力,容易适配业务需求。

import requests
from requests.exceptions import ConnectionError, Timeout, HTTPError

class FailoverSession(requests.Session):
    def __init__(self, target_list):
        super().__init__()
        self.targets = target_list  # 格式:["http://192.168.1.100", "http://device-02.example.com"]
        self.last_working_target = None  # 记录最近可用设备,优化后续请求

    def request(self, method, endpoint, **kwargs):
        # 优先尝试最近成功的设备,减少无效遍历
        attempt_sequence = []
        if self.last_working_target:
            attempt_sequence.append(self.last_working_target)
            attempt_sequence.extend(t for t in self.targets if t != self.last_working_target)
        else:
            attempt_sequence = self.targets.copy()

        for idx, target in enumerate(attempt_sequence):
            # 拼接完整URL,处理路径首尾的斜杠问题
            full_url = f"{target.rstrip('/')}/{endpoint.lstrip('/')}"
            try:
                resp = super().request(method, full_url, **kwargs)
                # 根据业务需求调整成功判定:这里以2xx状态码为准
                resp.raise_for_status()
                self.last_working_target = target
                return resp
            except (ConnectionError, Timeout, HTTPError) as e:
                # 最后一个设备尝试失败则抛出异常
                if idx == len(attempt_sequence) - 1:
                    raise RuntimeError(f"所有冗余设备均不可用: {str(e)}") from e
                # 跳过故障设备,尝试下一个
                continue

# 使用示例
if __name__ == "__main__":
    device_targets = [
        "http://192.168.1.100",
        "http://192.168.1.101",
        "dc-hardware-03.example.com"
    ]
    session = FailoverSession(device_targets)

    try:
        # 只需传入API端点,无需指定完整URL
        response = session.get("/api/system/status", timeout=5)
        print(f"成功连接至: {response.url}")
        print(f"设备状态: {response.json()}")
    except RuntimeError as e:
        print(f"请求失败: {e}")

方案优势

  • 无需依赖第三方库,完全基于requests原生API
  • 可灵活调整成功判定逻辑(比如允许特定非2xx状态码、自定义业务错误码)
  • 内置最近可用设备缓存,减少无效尝试次数
  • 支持自定义超时、重试等参数(通过kwargs传递)

方案2:底层urllib3适配器控制(进阶)

如果需要更底层的连接池控制,可以自定义urllib3.HTTPAdapter,在连接建立阶段实现目标切换。不过这种方式复杂度较高,仅适合需要精细控制连接行为的场景,大部分业务需求用方案1即可覆盖。

优化建议

  • 添加设备健康检查缓存:标记故障设备,间隔一段时间后再重试,避免频繁请求已失效节点
  • 配置超时阈值:单个设备请求超时时间不宜过长,避免阻塞流程
  • 支持负载均衡策略:比如轮询、权重分配(在attempt_sequence中调整顺序即可实现)

内容的提问来源于stack exchange,提问作者Andreas Maier

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.22 13:15:26