Python HTTP 客户端实践：超时、有限重试与幂等边界

📅 2026-10-21 📂 后端 🏷️ Python, HTTP, urllib, 重试, 幂等
12 分钟阅读 · 2,727 字

后端请求失败时，最容易写出的处理是“再试一次”。但响应超时不代表服务端没有处理请求；盲目重放一次付款、创建任务或发送消息，可能造成重复副作用。可靠的 HTTP 客户端需要先把请求语义讲清楚，再设置总尝试次数、单次超时、总耗时预算、响应体上限和退避策略。

本文使用 Python 3.9+ 标准库 urllib，只实现只读 JSON GET。示例不会访问真实 API，测试使用可控的假响应。生产系统若已有 httpx、requests 等客户端，应优先遵循项目现有依赖与封装，不要同时叠加多套重试层。这里的 3.9 是示例兼容下限，不是新项目的安全建议：Python 3.9 已于 2025-10-31 结束支持；请选择官方仍维护的版本，见 Python 3.9.25 发布说明。

1. 先区分 HTTP 错误和传输错误

urllib.request.urlopen() 成功时返回可关闭的响应对象；HTTP 错误通常抛出 HTTPError，它包含状态码和响应头，也有文件对象接口；DNS、连接等问题通常表现为 URLError，底层超时还可能以 TimeoutError 出现。因此异常处理要先捕获 HTTPError，再处理连接类错误，不能把所有异常都当成“可以重试”。

请求 URL 也要有边界。下面的例子拒绝非 HTTP(S) 地址、空主机名和 URL 内嵌用户名密码。它适用于由配置提供的 API 地址，不是任意 URL 抓取器；服务器若读取用户提交的 URL，还必须做主机白名单、DNS/私网地址限制，并在每次重定向后重新校验目标，以防 SSRF。

2. 为只读请求设置有界重试

这段实现固定为 GET，并限制最多 5 次尝试、最多 60 秒的整体预算、每次最多 3 秒、64 KiB 响应体和最多 30 秒的服务端 Retry-After 等待。网络失败与少数暂时性状态码才进入退避；鉴权、参数、权限和不存在等错误直接返回调用方。

import json
import math
import random
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

MAX_BODY_BYTES = 64 * 1024
ATTEMPT_TIMEOUT_SECONDS = 3.0
MAX_ATTEMPTS = 5
MAX_TOTAL_TIMEOUT_SECONDS = 60.0
BASE_BACKOFF_SECONDS = 0.25
MAX_BACKOFF_SECONDS = 2.0
MAX_RETRY_AFTER_SECONDS = 30.0
RETRYABLE_STATUS = frozenset({408, 429, 500, 502, 503, 504})


def validate_api_url(url):
    if not isinstance(url, str) or not url:
        raise ValueError("API URL 必须是非空字符串")
    parts = urlsplit(url)
    try:
        port = parts.port
    except ValueError as error:
        raise ValueError("API URL 端口无效") from error
    if (
        parts.scheme.lower() not in {"http", "https"}
        or not parts.hostname
        or parts.username is not None
        or parts.password is not None
        or (port is not None and not 1 <= port <= 65535)
    ):
        raise ValueError("API URL 必须是有效的 HTTP(S) 地址，且不能包含账号密码")
    return url


def parse_retry_after(value, *, now=None):
    """解析 Retry-After 秒数或 HTTP-date；无法解析时返回 None。"""
    if not isinstance(value, str):
        return None
    value = value.strip()
    if not value:
        return None
    if value.isascii() and value.isdigit():
        return float(int(value))

    try:
        target = parsedate_to_datetime(value)
    except (TypeError, ValueError, OverflowError):
        return None
    if target is None:
        return None
    if target.tzinfo is None:
        target = target.replace(tzinfo=timezone.utc)
    current_epoch = time.time() if now is None else now
    current = datetime.fromtimestamp(current_epoch, timezone.utc)
    return max(0.0, (target - current).total_seconds())


def fetch_json_get(
    url,
    *,
    timeout=8.0,
    max_attempts=3,
    opener=None,
    sleep=None,
    monotonic=None,
    wall_time=None,
    rng=None,
):
    """读取 JSON GET；注入时钟和 opener 以便离线测试。"""
    validate_api_url(url)
    if isinstance(timeout, bool) or not isinstance(timeout, (int, float)):
        raise ValueError("timeout 必须是有限正数")
    if not math.isfinite(timeout) or not 0 < timeout <= MAX_TOTAL_TIMEOUT_SECONDS:
        raise ValueError("timeout 必须在 0 到 60 秒之间")
    if type(max_attempts) is not int or not 1 <= max_attempts <= MAX_ATTEMPTS:
        raise ValueError("max_attempts 必须是 1 到 5 之间的整数")

    open_request = opener or urlopen
    sleep_for = sleep or time.sleep
    clock = monotonic or time.monotonic
    current_wall_time = wall_time or time.time
    random_value = rng or random.random
    request = Request(url, headers={"Accept": "application/json"}, method="GET")
    deadline = clock() + float(timeout)

    for attempt in range(1, max_attempts + 1):
        remaining = deadline - clock()
        if remaining <= 0:
            raise TimeoutError("HTTP 请求总耗时预算已用完")

        retry_after = None
        last_error = None
        try:
            with open_request(request, timeout=min(remaining, ATTEMPT_TIMEOUT_SECONDS)) as response:
                body = response.read(MAX_BODY_BYTES + 1)
            if len(body) > MAX_BODY_BYTES:
                raise ValueError("HTTP 响应超过 64 KiB 限制")
            return json.loads(body.decode("utf-8"))
        except HTTPError as error:
            last_error = error
            headers = getattr(error, "headers", None)
            header_value = headers.get("Retry-After") if headers else None
            retry_after = parse_retry_after(header_value, now=current_wall_time())
            retryable = error.code in RETRYABLE_STATUS
            error.close()
            if not retryable or attempt == max_attempts:
                raise
        except (URLError, TimeoutError) as error:
            last_error = error
            if attempt == max_attempts:
                raise

        if retry_after is not None:
            if retry_after > MAX_RETRY_AFTER_SECONDS:
                raise last_error
            delay = retry_after
        else:
            backoff_cap = min(MAX_BACKOFF_SECONDS, BASE_BACKOFF_SECONDS * (2 ** (attempt - 1)))
            sample = max(0.0, min(1.0, float(random_value())))
            delay = sample * backoff_cap

        remaining = deadline - clock()
        if delay >= remaining:
            raise TimeoutError("等待重试会超过 HTTP 请求总耗时预算") from last_error
        sleep_for(delay)

    raise AssertionError("重试循环不应落到这里")

配置与选择有几条值得明确：

  • HTTPError 是 URLError 的子类，所以代码必须先处理 HTTPError，才能按状态码和 Retry-After 决定是否重试。
  • 退避采用指数上限加随机抖动，减少多个客户端同时恢复时形成的重试尖峰；尝试次数仍是硬上限。
  • Retry-After 可以是秒数或 HTTP 日期。若服务端要求等待超过本地策略允许的 30 秒，示例会返回原错误，不会擅自缩短等待后提前重试。
  • 响应读取 limit + 1 字节，超过大小立即失败，不把任意大的 JSON 一次性读进内存。
  • time.monotonic() 用于测量耗时预算，避免系统时钟校准影响差值；time.time() 只用于把 HTTP 日期转换成等待秒数。
  • 这个预算不能中断正在执行的阻塞读取。如果上游持续逐字节发送数据，单次读取可能越过总预算；urllib 的 timeout 不是严格的端到端 deadline。对硬截止时间、异步取消或复杂连接池需求，应使用项目支持的客户端和可取消任务边界，并单独测量行为。

3. 在调用层处理可预期失败

try:
    profile = fetch_json_get(
        "https://api.example.test/v1/profile",
        timeout=5.0,
        max_attempts=3,
    )
except (HTTPError, URLError, TimeoutError, ValueError) as error:
    # 写入脱敏错误类别；不要记录授权头、完整响应正文或敏感查询参数。
    print("读取 profile 失败：", type(error).__name__)
else:
    print("返回字段数：", len(profile) if isinstance(profile, dict) else "非对象响应")

示例中的 URL 是文档占位域名，代码不会在文章加载时运行。真实项目通常会把错误转换成稳定的业务错误类型，由上层决定展示、降级或结束任务；不要在每一层各自再套一层重试，否则 API 客户端、任务队列、网关的次数会相乘。

4. 用假响应验证重试协议

以下测试不联网，验证请求方法、响应大小、重试次数、指数退避和 Retry-After 上界。将第 2 节代码与测试放在同一文件即可运行。

import io
import unittest


class FakeResponse:
    def __init__(self, body, status=200):
        self.body = body
        self.status = status
        self.read_limit = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, traceback):
        return False

    def getcode(self):
        return self.status

    def read(self, limit):
        self.read_limit = limit
        return self.body[:limit]


def http_error(status, headers=None):
    return HTTPError(
        "https://api.example.test/data",
        status,
        "temporary failure",
        headers or {},
        io.BytesIO(b"not logged"),
    )


class HttpClientGuideTests(unittest.TestCase):
    def test_success_is_a_single_bounded_get(self):
        response = FakeResponse(b'{"ok": true}')
        calls = []

        def opener(request, timeout):
            calls.append((request.get_method(), timeout))
            return response

        result = fetch_json_get("https://api.example.test/data", opener=opener)
        self.assertEqual(result, {"ok": True})
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0][0], "GET")
        self.assertLessEqual(calls[0][1], ATTEMPT_TIMEOUT_SECONDS)
        self.assertEqual(response.read_limit, MAX_BODY_BYTES + 1)

    def test_503_honors_retry_after_seconds(self):
        clock_state = {"now": 100.0}
        delays = []
        attempts = iter([http_error(503, {"Retry-After": "1"}), FakeResponse(b'{"ready": true}')])

        def opener(request, timeout):
            result = next(attempts)
            if isinstance(result, Exception):
                raise result
            return result

        def sleep(delay):
            delays.append(delay)
            clock_state["now"] += delay

        result = fetch_json_get(
            "https://api.example.test/data",
            timeout=5,
            max_attempts=2,
            opener=opener,
            sleep=sleep,
            monotonic=lambda: clock_state["now"],
            wall_time=lambda: 100.0,
        )
        self.assertEqual(result, {"ready": True})
        self.assertEqual(delays, [1.0])

    def test_unavailable_host_stops_at_the_attempt_limit(self):
        clock_state = {"now": 0.0}
        calls = []
        delays = []

        def opener(request, timeout):
            calls.append(timeout)
            raise URLError("connection reset")

        def sleep(delay):
            delays.append(delay)
            clock_state["now"] += delay

        with self.assertRaises(URLError):
            fetch_json_get(
                "https://api.example.test/data",
                timeout=5,
                max_attempts=2,
                opener=opener,
                sleep=sleep,
                monotonic=lambda: clock_state["now"],
                rng=lambda: 0.5,
            )
        self.assertEqual(len(calls), 2)
        self.assertEqual(len(delays), 1)
        self.assertLessEqual(delays[0], BASE_BACKOFF_SECONDS)

    def test_oversized_body_fails_without_retry(self):
        calls = []

        def opener(request, timeout):
            calls.append(request.get_method())
            return FakeResponse(b"x" * (MAX_BODY_BYTES + 1))

        with self.assertRaisesRegex(ValueError, "64 KiB"):
            fetch_json_get("https://api.example.test/data", opener=opener)
        self.assertEqual(calls, ["GET"])

    def test_long_retry_after_is_not_shortened_to_an_early_retry(self):
        calls = []
        delays = []

        def opener(request, timeout):
            calls.append(request.get_method())
            raise http_error(503, {"Retry-After": "31"})

        with self.assertRaises(HTTPError):
            fetch_json_get(
                "https://api.example.test/data",
                opener=opener,
                sleep=delays.append,
                wall_time=lambda: 100.0,
            )
        self.assertEqual(calls, ["GET"])
        self.assertEqual(delays, [])

    def test_url_and_retry_after_date_validation(self):
        for url in (
            "file:///etc/passwd",
            "https://user:pass@api.example.test/data",
            "https:///missing-host",
        ):
            with self.subTest(url=url), self.assertRaises(ValueError):
                validate_api_url(url)
        self.assertEqual(parse_retry_after("Thu, 01 Jan 1970 00:16:41 GMT", now=1000), 1.0)
        self.assertIsNone(parse_retry_after("not a date", now=1000))


if __name__ == "__main__":
    unittest.main()

运行：

python -m unittest -v http_client_guide.py

5. 哪些请求不应自动重放

RFC 9110 将幂等定义为：相同请求重复执行，服务端预期效果与执行一次相同。GET、HEAD、PUT、DELETE 按标准语义属于幂等方法，但具体 API 仍可能实现额外副作用，必须核对接口契约。POST 通常不具备幂等语义；连接断开或客户端超时后，服务端可能已经完成写入，只是响应没有到达客户端。

若业务需要重试创建类 POST，必须由服务端提供并持久化幂等键，且让同一个业务操作的所有重试使用同一个键；还要定义键的作用域、保留期、并发请求竞争、请求体冲突和原始结果重放。仅仅发送 Idempotency-Key 这个请求头，如果服务端没有实现去重契约，不会自动变安全。支付、发消息、创建推理任务等流程还需要查询状态、任务 ID 或人工补偿路径。

把边界记成一句话：超时说明客户端没有及时拿到结果，不说明服务端没有执行。 先设计幂等和结果查询，再讨论重试；否则失败重放会把一次不确定操作变成重复操作。

资料

本文是原创学习指南，不替代具体 API 的服务端契约、SLO、隐私规范或项目依赖策略。更完整的 Python 学习次序见 Python 学习路线；异步任务的取消、期限和结构化并发见 asyncio 任务管理指南。

Live Notes

读者弹幕

留言提交后会先进入审核，审核通过后对其他读者公开。