Python 基础体系 · 第 60/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。
Python HTTP 客户端：连接池、超时、重试、流式和 TLS
HTTP 客户端不是“调用一个 URL 并读取 JSON”这么简单。一次请求至少可能经历 DNS 解析、TCP 建连、TLS 握手、连接池排队、发送请求头、上传请求体、等待响应头、持续读取响应体，以及关闭或复用连接等阶段。
如果这些阶段没有被分别建模，工程中常见的问题就很难解释：
- 请求偶尔卡住，但设置了超时；
- 并发一上升，延迟突然变高；
- 重试后出现重复扣款或重复创建资源；
- 读取大文件时内存暴涨；
- HTTPS 在开发环境正常，在生产环境证书校验失败；
- FastAPI 中使用异步客户端，却因为客户端生命周期错误导致连接池失效。
本文以 Python 3.14 为运行环境，使用标准库解释底层机制，并以 HTTPX 展示实际工程中的同步和异步写法。HTTPX 不是 Python 标准库的一部分，需要单独安装；它提供同步、异步、HTTP/1.1、HTTP/2、连接池、流式响应和 TLS 配置等能力。(python-httpx.org)
一、先建立完整的请求模型
1. 一次 HTTP 请求经过哪些阶段
把客户端请求抽象成以下状态：
准备请求
│
├─ 连接池中有可复用连接 ─────────────┐
│ │
└─ 没有可用连接 │
│ │
├─ 等待连接池槽位 │
├─ DNS 解析 │
├─ TCP connect │
├─ TLS handshake（HTTPS） │
└─ 建立协议连接 │
▼
发送请求头和请求体
│
▼
接收响应头
│
▼
读取响应体
│
┌─────────────────┴─────────────────┐
│ │
完整读取并复用 出错或关闭
│ │
▼ ▼
归还连接池 关闭连接
这里的“连接”不是一个抽象的 URL，而是底层 TCP 连接，HTTPS 下还包括该 TCP 连接之上的 TLS 会话。
对于 HTTP/1.1，一条连接在响应体尚未读取完之前，通常不能安全地交给另一个请求使用。于是：
连接池中的连接数 ≠ 当前正在创建的请求数
更准确地说：
max_connections限制同时占用的连接总数；max_keepalive_connections限制空闲、可复用连接的数量；- 正在读取响应体的请求会继续占用连接；
- 响应体没有读取完就放弃，客户端通常只能关闭连接，而不能复用它。
HTTPX 的 Client 会复用底层 TCP 连接，而每次直接调用 httpx.get() 这样的顶层 API 则不会像长期存在的 Client 一样提供连接池复用。(python-httpx.org)
2. HTTP、TCP、TLS 的分层关系
HTTP 客户端通常位于以下层次之上：
应用代码
│ JSON、状态码、分页、幂等键
HTTP
│ 请求行、响应头、响应体
TLS（HTTPS 时存在）
│ 加密、身份认证、完整性保护
TCP
│ 可靠、有序的字节流
IP
│ 寻址和路由
TCP 提供的是字节流，不是消息。HTTP/1.1 需要依靠请求头、Content-Length、分块传输编码等机制判断一条消息的边界；HTTP 客户端负责把这些字节流解释成请求和响应。
TLS 也不负责定义 HTTP。TLS 在可靠的字节流之上建立安全通道，提供服务器认证、机密性和完整性；握手阶段协商参数并建立密钥，后续记录协议使用这些密钥保护应用数据。(rfc-editor.org)
二、标准库能做什么，HTTPX 解决什么问题
1. 使用 http.client 直接复用连接
Python 标准库的 http.client 是低级 HTTP 客户端，urllib.request 也通过它处理 HTTP 和 HTTPS。(docs.python.org)
最底层的连接复用可以这样观察：
from http.client import HTTPSConnection
conn = HTTPSConnection("example.com", timeout=5)
try:
conn.request("GET", "/")
response = conn.getresponse()
body = response.read() # 必须读完，才能尝试复用连接
print(response.status)
print(body[:100])
conn.request("GET", "/")
response = conn.getresponse()
print(response.status)
response.read()
finally:
conn.close()
这个例子中，第二次请求有机会复用第一次请求创建的连接。关键条件是：
response.read()
如果响应体没有读完，客户端不知道连接中的字节是否已经属于下一条响应，也不能安全地把连接交给后续请求。Python 文档明确要求：在同一个 HTTPConnection 上发起下一次请求前，应先读完整个响应，或者关闭连接。(docs.python.org)
2. 为什么工程中通常使用客户端对象
低级 API 需要调用者自己处理：
- 不同主机的连接管理；
- 并发访问时的锁和连接分配；
- 空闲连接回收；
- 连接池大小；
- 超时分类；
- 流式响应关闭；
- 重试和退避；
- 代理、TLS 和 HTTP/2。
HTTPX 将这些能力封装到 Client 和 AsyncClient 中。工程代码中，客户端通常应当拥有比单次函数调用更长的生命周期：
import httpx
with httpx.Client(base_url="https://api.example.com") as client:
response = client.get("/users")
response.raise_for_status()
users = response.json()
这里的 with 不只是语法简化，它保证退出作用域时关闭连接池。长期运行的服务可以把客户端放在应用生命周期中，而不是每次请求都创建一个新客户端。
错误的写法是把客户端放进热循环：
for user_id in user_ids:
with httpx.Client() as client:
client.get(f"https://api.example.com/users/{user_id}")
这会反复创建和销毁连接池，连接复用的价值基本消失。HTTPX 的异步文档也特别提醒，不要在热循环中不断创建 AsyncClient，否则无法获得有效的连接池复用。(python-httpx.org)
三、连接池：复用连接，而不是无限增加并发
1. 连接池中的三个数量
以 HTTPX 为例：
import httpx
limits = httpx.Limits(
max_connections=100,
max_keepalive_connections=20,
keepalive_expiry=5.0,
)
client = httpx.Client(limits=limits)
三个参数含义不同：
| 参数 | 含义 |
|---|---|
max_connections |
同时允许占用的连接总数 |
max_keepalive_connections |
空闲连接最多保留多少条 |
keepalive_expiry |
空闲连接保留多久 |
HTTPX 当前文档给出的默认值是 max_connections=100、max_keepalive_connections=20、keepalive_expiry=5 秒；实际使用时应以所安装版本的文档和运行行为为准。(python-httpx.org)
可以用一个简单模型理解连接池：
设：
C为最大连接数；A为当前正在使用的连接数；K为池中空闲可复用连接数；W为正在等待连接的请求数。
则通常有：
0 ≤ A ≤ C
0 ≤ K ≤ max_keepalive_connections
W > 0 ⇔ 没有空闲连接且 A 已达到 C
并发请求数超过 C 时，并不是自动创建更多连接，而是进入等待状态。等待本身也必须有超时，否则连接池耗尽时请求可能长期阻塞。
2. 连接池不是全局限流器
连接池限制的是连接资源，不一定等价于业务并发限制。
例如：
最大连接数 C = 10
每个请求平均持续时间 S = 2 秒
在理想情况下，吞吐上限大致受：
吞吐量 ≈ C / S = 10 / 2 = 5 请求/秒
影响。
这个公式只是粗略估算，因为真实请求还会受到服务端处理、网络带宽、响应体大小、连接复用、HTTP/2 多路复用等因素影响。但它能解释一个现象：把连接数从 10 增加到 100，不一定让系统变快，可能只是把压力转移到：
- 本机文件描述符；
- NAT 表；
- 服务端连接数；
- 数据库连接；
- CPU 上下文切换；
- 连接池等待和超时。
3. HTTP/1.1 与 HTTP/2 的差异
HTTP/1.1 中，一条 TCP 连接通常按请求响应顺序处理；多个并发请求往往需要多个连接。
HTTP/2 支持在一条连接上复用多个 stream，因此：
HTTP/1.1：并发请求通常消耗多个 TCP 连接
HTTP/2：多个请求可以共享一条 TCP 连接
但这不意味着连接数限制可以随意设为 1。HTTP/2 仍然受服务端最大并发 stream、流控窗口、带宽和单连接队头阻塞等因素影响。HTTPX 同时支持 HTTP/1.1 和 HTTP/2，但是否实际使用 HTTP/2 取决于客户端配置和服务端协商。(python-httpx.org)
四、超时：不是一个数字，而是一组阶段预算
1. 四种超时
HTTPX 将超时分成四类：
- connect timeout：建立连接时最多等待多久；
- read timeout：等待下一块响应数据最多多久；
- write timeout：发送下一块请求数据最多多久；
- pool timeout：从连接池获取连接最多等待多久。
如果连接池已满，触发的是 PoolTimeout，而不是 ConnectTimeout。如果连接已建立但服务端长时间不返回数据，触发的是 ReadTimeout。(python-httpx.org)
import httpx
timeout = httpx.Timeout(
connect=3.0,
read=10.0,
write=10.0,
pool=2.0,
)
with httpx.Client(timeout=timeout) as client:
response = client.get("https://api.example.com/items")
response.raise_for_status()
这段配置表达的是：
连接建立最多等 3 秒
每次读取数据最多等 10 秒
每次写入数据最多等 10 秒
等待池中连接最多等 2 秒
2. 读取超时不是整个响应的总时长
假设：
read timeout = 5 秒
服务端每隔 4 秒发送一个字节，客户端可能一直不超时，因为每次等待下一块数据都没有超过 5 秒。
因此：
read timeout ≠ 整个请求的 deadline
如果业务要求“整个请求最多只能运行 30 秒”，还需要在应用层设置总截止时间：
import time
import httpx
deadline = time.monotonic() + 30.0
try:
with httpx.Client(timeout=httpx.Timeout(5.0)) as client:
response = client.get("https://api.example.com/large-result")
response.raise_for_status()
except httpx.TimeoutException as exc:
print(f"阶段超时: {type(exc).__name__}")
在更复杂的重试场景中，应让每一次重试共享同一个总预算，而不是每次都重新获得完整的 30 秒：
总预算 T = 30 秒
第 1 次请求消耗 8 秒
退避消耗 2 秒
第 2 次请求最多只能再消耗 20 秒
否则：
每次请求 30 秒 × 3 次 = 最坏 90 秒
会超过调用方对整个操作的预期。
3. 为什么不能无限关闭超时
下面的代码会允许网络操作永久等待：
import httpx
with httpx.Client(timeout=None) as client:
client.get("https://api.example.com")
HTTPX 默认会对网络不活动进行超时控制，默认网络不活动时间为 5 秒；可以通过 timeout 修改或显式关闭。(python-httpx.org)
timeout=None 只适合非常明确的场景，例如：
- 由外部 watchdog 控制生命周期；
- 专门处理长连接；
- 上层有更严格的总 deadline；
- 能够正确取消任务和释放连接。
普通 API 调用关闭超时，会把单个异常请求变成长期占用连接池槽位的请求，最终表现为大量 PoolTimeout。
五、重试：先判断“请求是否可能已经成功”
1. 网络错误不等于服务端没有执行
考虑以下时序：
客户端 ── POST 创建订单 ──> 服务端
客户端 <── 服务端已写入订单
网络连接断开
客户端没有收到响应
客户端看到的是连接错误，但服务端可能已经成功执行了请求。如果客户端直接重试：
第一次：订单已创建，但响应丢失
第二次：再次创建订单
就可能产生重复订单。
因此，重试前必须区分两个问题：
- 请求是否可以安全重复；
- 第一次请求是否可能已经被服务端应用。
HTTP 规范将幂等方法定义为：重复执行多个相同请求，其预期的服务器效果与执行一次相同。幂等性之所以重要，是因为客户端可能在无法读取响应时自动重试；对于非幂等请求，客户端不应在没有额外保证的情况下自动重试。(rfc-editor.org)
2. 方法名不等于业务幂等性
常见方法的语义倾向如下：
| 方法 | 通常是否幂等 | 说明 |
|---|---|---|
GET |
是 | 获取资源，不应产生业务副作用 |
HEAD |
是 | 获取元信息 |
PUT |
是 | 将资源设置为指定表示 |
DELETE |
是 | 删除目标资源，重复删除通常仍是删除状态 |
POST |
通常否 | 可能每次创建新资源 |
PATCH |
取决于设计 | 可能是增量修改，也可能是幂等设置 |
但方法名只是默认语义。一个 POST /payments 可以通过幂等键设计成业务幂等：
POST /payments
Idempotency-Key: 2c5e3c4d-...
服务端需要持久化这个键与最终结果的映射：
第一次收到 key=K：
执行支付
保存 K -> 支付结果
再次收到 key=K：
不再次扣款
返回之前保存的结果
只有服务端真的实现了这种语义，客户端才有理由对该 POST 进行受控重试。
3. 哪些错误值得重试
通常可以考虑重试：
- 连接建立失败；
- 连接建立超时；
- 临时 DNS 或网络错误；
502 Bad Gateway；503 Service Unavailable；504 Gateway Timeout；- 明确由服务端约定为临时故障的响应。
通常不应盲目重试：
400 Bad Request；401 Unauthorized；403 Forbidden；404 Not Found；- 参数校验错误；
- TLS 证书校验失败；
- 请求体不可重复读取；
- 非幂等操作没有幂等键。
HTTP 503 响应可以携带 Retry-After，表示客户端应等待多久再发起后续请求；该字段可以是秒数，也可以是 HTTP 日期。(rfc-editor.org)
4. 指数退避和抖动
简单的指数退避可以写成：
delay_n = min(cap, base × 2^n)
其中：
n是已经失败的重试次数，从 0 开始；base是基础等待时间；cap是最大等待上限。
例如：
base = 0.5 秒
cap = 8 秒
第 1 次重试：0.5 秒
第 2 次重试：1 秒
第 3 次重试：2 秒
第 4 次重试：4 秒
第 5 次重试：8 秒
如果所有客户端同时失败并按照完全相同的时间重试，会形成“惊群”。因此应加入随机抖动：
delay = random(0, min(cap, base × 2^n))
下面是一个仅对幂等请求和临时状态进行重试的同步示例：
import random
import time
import httpx
RETRYABLE_STATUS = {502, 503, 504}
def get_with_retry(
client: httpx.Client,
url: str,
*,
attempts: int = 3,
) -> httpx.Response:
last_error: Exception | None = None
for retry_index in range(attempts):
try:
response = client.get(url)
if response.status_code not in RETRYABLE_STATUS:
response.raise_for_status()
return response
# 先读取或关闭当前响应，避免连接被长期占用。
retry_after = response.headers.get("Retry-After")
response.close()
if retry_index == attempts - 1:
raise httpx.HTTPStatusError(
f"retryable status remains: {response.status_code}",
request=response.request,
response=response,
)
if retry_after and retry_after.isdigit():
delay = min(float(retry_after), 10.0)
else:
delay = random.uniform(
0.0,
min(10.0, 0.5 * (2**retry_index)),
)
time.sleep(delay)
except (httpx.ConnectError, httpx.ConnectTimeout) as exc:
last_error = exc
if retry_index == attempts - 1:
raise
delay = random.uniform(
0.0,
min(10.0, 0.5 * (2**retry_index)),
)
time.sleep(delay)
raise RuntimeError("unreachable") from last_error
这个示例只对 GET 使用自动重试，因此不涉及请求体重复发送。HTTPX 的 HTTPTransport(retries=n) 只覆盖连接错误和连接超时；如果还要根据 503、读取错误或其他状态码重试，需要在更高层实现策略，或使用专门的重试库。(python-httpx.org)
5. 重试的真实边界
下面这些情况不能简单地套用“失败就重试”：
响应已经返回，但业务处理失败
HTTP 200
响应 JSON：{"status": "processing"}
这不是传输失败，而是业务状态未完成。应该根据业务协议轮询、订阅事件或查询状态，而不是重发原请求。
请求体不可重复读取
def body():
yield from open("large-file.bin", "rb")
如果第一次发送已经消耗了部分生成器，第二次重试时可能没有完整请求体。可重试请求必须具备：
- 可重复读取的内存数据；
- 可重新打开的文件；
- 可重新生成且结果一致的请求体；
- 或者明确的断点续传协议。
超时发生在响应阶段
ReadTimeout 只能说明客户端在某个读取阶段没有及时收到数据，不能说明服务端没有执行请求。对 POST 的读取超时尤其危险，仍然需要幂等键或状态查询。
六、流式请求和响应：让内存、连接和处理速度保持一致
1. 非流式读取的内存模型
如果响应体大小为 N 字节，非流式代码通常会把完整响应体读入内存：
import httpx
with httpx.Client() as client:
response = client.get("https://example.com/archive.zip")
response.raise_for_status()
data = response.content
粗略地说，进程至少需要为响应体保留接近 N 的内存，解析过程还可能产生额外副本：
网络缓冲 + response.content + 解压缓冲 + 业务对象
如果 N = 2 GiB，显然不适合一次性读入内存。
2. 流式响应
HTTPX 提供上下文管理器来保证流式响应最终关闭：
import httpx
with httpx.Client() as client:
with client.stream(
"GET",
"https://example.com/archive.zip",
) as response:
response.raise_for_status()
with open("archive.zip", "wb") as output:
for chunk in response.iter_bytes(chunk_size=64 * 1024):
output.write(chunk)
数据流路径是：
服务端
│
▼
socket 接收缓冲
│
▼
HTTPX 迭代器
│
▼
应用每次读取 64 KiB
│
▼
文件系统
iter_bytes() 每次只交付一部分响应体，因此应用不必同时保存整个文件。更重要的是，with client.stream(...) 退出时会关闭或释放响应，避免连接池中的连接永久处于占用状态。
异步版本如下：
import asyncio
from pathlib import Path
import httpx
async def download(url: str, destination: Path) -> None:
timeout = httpx.Timeout(
connect=5.0,
read=30.0,
write=30.0,
pool=5.0,
)
async with httpx.AsyncClient(timeout=timeout) as client:
async with client.stream("GET", url) as response:
response.raise_for_status()
with destination.open("wb") as output:
async for chunk in response.aiter_bytes(64 * 1024):
output.write(chunk)
await asyncio.sleep(0)
HTTPX 的异步流式 API 使用 async with client.stream(...) 和 aiter_bytes()；异步客户端应在应用生命周期结束时通过 aclose() 或异步上下文管理器释放连接。(python-httpx.org)
3. 流式读取中的背压
背压是指下游处理速度限制上游数据读取速度。
例如，服务端以每秒 10 MiB 发送数据，而本地磁盘只能每秒写入 2 MiB：
服务端发送速度：10 MiB/s
磁盘写入速度： 2 MiB/s
如果客户端不断把数据读进内存，内存中的待写数据会增长。流式处理的正确目标不是“边读边写”这么简单，而是让读取速度受到下游容量限制：
读取一块
→ 写入磁盘
→ 写入完成后再读取下一块
异步代码中，如果下游是阻塞的普通文件写入，应注意它会短暂阻塞事件循环。对高吞吐服务，可以使用异步文件库、线程池或专门的对象存储 SDK；关键不是盲目增加 chunk 大小，而是使生产、传输和消费三个阶段有明确的容量边界。
4. 流式请求体
上传大文件时，也不应把文件全部读入内存：
import httpx
with open("large-file.bin", "rb") as file:
with httpx.Client() as client:
response = client.post(
"https://api.example.com/upload",
content=file,
)
response.raise_for_status()
这里请求体是文件对象，客户端可以逐步读取并发送。风险在于：
- 发送失败后文件指针可能已经移动；
- 自动重试需要重新打开文件或执行
seek(0)； - 服务端是否接受分块传输需要确认；
- 如果服务端要求
Content-Length，应确保客户端能够确定文件大小。
因此，大文件上传的重试设计通常比下载更复杂。
七、TLS：HTTPS 不只是把 URL 改成 https://
1. TLS 解决什么问题
TLS 主要提供三类安全属性：
- 认证：客户端验证自己连接的服务器身份；
- 机密性：网络观察者不能直接读取应用数据；
- 完整性：网络攻击者修改数据时能够被检测。
TLS 1.3 的握手负责协商密码参数、认证通信端点并建立共享密钥；握手完成后，记录协议使用这些密钥保护应用数据。(rfc-editor.org)
HTTPS 的实际路径是：
DNS
↓
TCP connect
↓
TLS ClientHello / ServerHello
↓
证书链验证、主机名验证
↓
密钥建立
↓
HTTP 请求和响应
因此，HTTPS 连接耗时至少包含：
总耗时 =
连接池等待
+ DNS
+ TCP 建连
+ TLS 握手
+ 发送请求
+ 等待响应
+ 读取响应体
连接池复用的价值之一，就是减少后续请求重复进行 TCP 和 TLS 握手。
2. 默认进行证书和主机名校验
HTTPX 默认验证 HTTPS 连接的身份；证书无效、过期或主机名不匹配时，会抛出 SSL 相关连接错误。(python-httpx.org)
import httpx
try:
with httpx.Client() as client:
response = client.get("https://api.example.com")
response.raise_for_status()
except httpx.ConnectError as exc:
print(f"TLS 或底层连接失败: {exc}")
不要用下面的方式“修复”生产环境证书错误：
with httpx.Client(verify=False) as client:
client.get("https://api.example.com")
verify=False 会关闭证书验证，使客户端无法确认对端是不是目标服务器。它可以用于隔离问题的临时实验，但不应作为生产配置。
3. 使用自定义 CA
企业内部服务经常使用私有 CA。正确做法是把该 CA 加入验证上下文：
import ssl
import httpx
context = ssl.create_default_context(
cafile="/etc/my-company/ca-bundle.pem"
)
with httpx.Client(verify=context) as client:
response = client.get("https://internal-api.example")
response.raise_for_status()
这里的文件应当是签发服务端证书的信任根或中间 CA，而不是随意下载的服务端证书。信任范围过大，会扩大客户端接受的证书集合；信任范围过小，则会导致合法服务连接失败。
HTTPX 支持把标准库 ssl.SSLContext 传给客户端，也支持通过 SSL_CERT_FILE 和 SSL_CERT_DIR 指定证书来源。(python-httpx.org)
4. 双向 TLS：mTLS
普通 HTTPS 通常只验证服务器：
客户端验证服务器
服务器不要求客户端证书
双向 TLS 还要求客户端提供证书：
import ssl
import httpx
context = ssl.create_default_context(
cafile="/etc/my-company/ca-bundle.pem"
)
context.load_cert_chain(
certfile="/etc/my-company/client.crt",
keyfile="/etc/my-company/client.key",
)
with httpx.Client(verify=context) as client:
response = client.get("https://mtls.internal.example")
response.raise_for_status()
mTLS 的关键关系是：
- 客户端证书证明客户端拥有对应私钥；
- 服务端需要信任签发客户端证书的 CA；
- 客户端仍然需要验证服务端证书；
- 证书文件权限必须限制，私钥不能进入日志、镜像层或普通用户可读目录。
TLS 规范中的 Certificate、CertificateVerify 和 Finished 消息共同参与证书认证、握手签名和握手完整性确认。(rfc-editor.org)
5. SNI、主机名和代理
HTTPS 证书校验不仅看证书是否由可信 CA 签发，还要看证书中的主机名是否匹配请求目标。
因此，下面两种情况并不等价：
访问 https://api.example.com
访问 https://192.0.2.10
即使 IP 地址最终指向同一台机器，证书也可能只包含 api.example.com，不包含 IP 地址。
通过 HTTP 代理访问 HTTPS 时，常见流程是：
客户端 ── TCP ──> 代理
客户端 ── CONNECT api.example.com:443 ──> 代理
代理建立到目标服务器的隧道
客户端在隧道中执行 TLS 握手
HTTPX 文档将这种方式称为 HTTP tunnel；HTTPS 请求通常先连接 HTTP 代理，再通过 CONNECT 建立到目标的隧道。(python-httpx.org)
八、同步和异步：改变等待方式，不改变 HTTP 语义
1. 同步客户端
同步客户端适合：
- 命令行脚本；
- 后台任务；
- 不运行事件循环的普通程序；
- 调用链本身是同步的服务。
import httpx
def fetch_user(user_id: int) -> dict:
with httpx.Client(
base_url="https://api.example.com",
timeout=httpx.Timeout(5.0),
) as client:
response = client.get(f"/users/{user_id}")
response.raise_for_status()
return response.json()
如果该函数频繁调用，客户端应提升到更长生命周期，而不是每次调用都创建：
import httpx
class UserAPI:
def __init__(self) -> None:
self.client = httpx.Client(
base_url="https://api.example.com",
timeout=httpx.Timeout(5.0),
)
def close(self) -> None:
self.client.close()
def fetch_user(self, user_id: int) -> dict:
response = self.client.get(f"/users/{user_id}")
response.raise_for_status()
return response.json()
生产代码中应把 close() 接入应用关闭生命周期，否则进程退出前可能留下未释放的资源。
2. 异步客户端
异步客户端适合事件循环中的 I/O 并发：
import asyncio
import httpx
async def fetch_one(
client: httpx.AsyncClient,
user_id: int,
) -> dict:
response = await client.get(f"/users/{user_id}")
response.raise_for_status()
return response.json()
async def main() -> None:
async with httpx.AsyncClient(
base_url="https://api.example.com",
timeout=httpx.Timeout(5.0),
) as client:
users = await asyncio.gather(
fetch_one(client, 1),
fetch_one(client, 2),
fetch_one(client, 3),
)
print(users)
asyncio.run(main())
这里的并发不意味着创建三个客户端，而是三个任务共享同一个客户端和连接池。客户端内部会根据连接池上限分配连接；如果并发任务多于可用连接，超出的任务会等待或触发 PoolTimeout。
3. 在 FastAPI 中管理客户端生命周期
对于 ASGI 应用，不应在每个接口调用中创建客户端。可以在应用生命周期中创建和关闭：
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http = httpx.AsyncClient(
base_url="https://api.example.com",
timeout=httpx.Timeout(
connect=3.0,
read=10.0,
write=10.0,
pool=2.0,
),
)
try:
yield
finally:
await app.state.http.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/proxy/users/{user_id}")
async def proxy_user(user_id: int, request: Request):
client: httpx.AsyncClient = request.app.state.http
try:
response = await client.get(f"/users/{user_id}")
response.raise_for_status()
return response.json()
except httpx.TimeoutException:
return {"error": "upstream timeout"}
except httpx.HTTPStatusError as exc:
return {
"error": "upstream status",
"status": exc.response.status_code,
}
except httpx.RequestError:
return {"error": "upstream network error"}
真实项目中通常应把异常转换为明确的 HTTP 响应，而不是直接返回普通字典；示例为了突出客户端错误分类，省略了具体的错误响应模型。
九、错误处理：区分传输错误、协议错误和业务错误
HTTP 客户端失败至少分为三层。
1. 传输层错误
请求没有得到可用的 HTTP 响应：
DNS 失败
TCP 失败
TLS 失败
连接池超时
读取超时
写入超时
HTTPX 的典型异常包括：
httpx.ConnectError
httpx.ConnectTimeout
httpx.ReadTimeout
httpx.WriteTimeout
httpx.PoolTimeout
httpx.NetworkError
这些异常继承自请求或传输错误层次，可以统一捕获，也可以按阶段分别处理。(python-httpx.org)
2. HTTP 协议层错误
服务器返回了 HTTP 响应，但响应本身不满足协议或客户端预期，例如：
- HTTP 状态行格式错误；
- 响应体无法按声明编码解码；
- 重定向次数过多；
- 连接中途断开；
- 响应流已经被关闭。
这类错误不是“没有响应”，而是“响应无法被正确解释”。
3. 业务层错误
服务器返回了合法 HTTP 响应，但业务结果是失败：
HTTP/1.1 422 Unprocessable Entity
Content-Type: application/json
或者：
HTTP/1.1 200 OK
{"code": "INSUFFICIENT_BALANCE"}
200 OK 不等于业务成功。客户端应同时检查：
response.raise_for_status()
payload = response.json()
if payload.get("code") != "OK":
raise RuntimeError(f"业务失败: {payload}")
如果 API 契约规定所有错误使用 4xx/5xx，则可以通过 raise_for_status() 处理 HTTP 层错误；如果业务错误被放进 2xx 响应，就必须按照响应模型检查。
十、一个可运行的端到端示例
下面的服务模拟：
- 一个慢接口；
- 一个临时返回
503的接口； - 一个流式响应接口。
1. 启动测试服务
安装依赖：
python -m pip install fastapi uvicorn httpx
创建 server.py：
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.get("/slow")
async def slow():
await asyncio.sleep(1)
return {"ok": True}
@app.get("/stream")
async def stream():
async def generate():
for index in range(5):
yield f"line-{index}\n".encode()
await asyncio.sleep(0.2)
return StreamingResponse(
generate(),
media_type="application/octet-stream",
)
启动：
uvicorn server:app --host 127.0.0.1 --port 8000
2. 客户端代码
创建 client.py：
import httpx
def main() -> None:
timeout = httpx.Timeout(
connect=1.0,
read=2.0,
write=2.0,
pool=1.0,
)
limits = httpx.Limits(
max_connections=10,
max_keepalive_connections=5,
keepalive_expiry=10.0,
)
with httpx.Client(
base_url="http://127.0.0.1:8000",
timeout=timeout,
limits=limits,
) as client:
response = client.get("/slow")
response.raise_for_status()
print("普通响应:", response.json())
with client.stream("GET", "/stream") as response:
response.raise_for_status()
print("流式响应:")
for chunk in response.iter_bytes():
print(repr(chunk))
if __name__ == "__main__":
main()
运行：
python client.py
预期输出类似：
普通响应: {'ok': True}
流式响应:
b'line-0\n'
b'line-1\n'
b'line-2\n'
b'line-3\n'
b'line-4\n'
每一步的原因是：
Timeout防止连接、读取、写入和连接池等待无限持续；Limits限制并发连接和空闲连接；Client让多个请求共享连接池；raise_for_status()将 4xx/5xx 转成异常；client.stream()不把完整响应体一次性加载到内存；with保证客户端和响应流最终关闭。
十一、在 ASGI 测试中绕过真实网络
ASGI 是异步服务器网关接口规范，用于定义 Python Web 应用与服务器之间的调用协议。测试 HTTP 客户端时，可以让 HTTPX 通过 ASGITransport 直接调用应用，而不经过 TCP、DNS 和真实 TLS。
FastAPI 官方文档使用 httpx.AsyncClient 和 ASGITransport 编写异步测试。(fastapi.tiangolo.com)
import pytest
from httpx import ASGITransport, AsyncClient
from server import app
@pytest.mark.anyio
async def test_slow():
transport = ASGITransport(app=app)
async with AsyncClient(
transport=transport,
base_url="http://test",
) as client:
response = await client.get("/slow")
assert response.status_code == 200
assert response.json() == {"ok": True}
这个测试验证的是：
HTTPX 请求构造
→ ASGI scope
→ FastAPI 路由
→ ASGI response messages
→ HTTPX Response
它不会验证真实网络中的：
- DNS；
- TCP 连接；
- TLS 证书；
- 代理；
- 服务端监听端口；
- 真实连接池行为。
因此，ASGITransport 适合应用级测试，但不能替代真实 HTTPS 和网络故障测试。
如果应用依赖启动和关闭事件，还需要显式管理 ASGI lifespan；FastAPI 文档指出，直接使用 AsyncClient 时不会自动触发生命周期事件。(fastapi.tiangolo.com)
十二、诊断：从现象反推故障阶段
1. 大量 PoolTimeout
可能路径：
请求到达客户端
→ 连接池没有空闲连接
→ 已达到 max_connections
→ 等待 pool timeout
优先检查：
- 是否每次调用都创建客户端；
- 是否有响应体未读取或未关闭；
- 是否下载、上传任务长期占用连接；
max_connections是否低于实际并发；- 上游响应时间是否突然增加。
不要一看到 PoolTimeout 就直接把连接数调大。连接泄漏和响应未关闭同样会造成池耗尽。
2. 大量 ReadTimeout
可能路径：
连接已建立
→ 请求已发送
→ 等待响应体下一块数据超时
检查：
- 服务端是否真的处理很慢；
- 响应是否是长轮询或流式接口；
read timeout是否被误当成总 deadline；- 上游是否已经返回部分数据；
- 重试该请求是否会产生重复副作用。
3. CERTIFICATE_VERIFY_FAILED
常见原因：
- 证书过期；
- 主机名与证书 SAN 不匹配；
- 客户端缺少内部 CA；
- 系统时间错误；
- 代理进行了 TLS 中间人解密，但客户端不信任代理 CA；
- 服务端未发送完整证书链。
诊断时可以临时查看证书链，但不要以关闭校验作为修复方案。最终应修复 CA、证书链、主机名或代理信任配置。
4. 重试后服务端压力更大
如果客户端重试间隔固定：
第 1 次失败：所有客户端等待 1 秒
第 2 次失败：所有客户端等待 2 秒
第 3 次失败：所有客户端等待 4 秒
大量客户端会在相同时间再次冲击上游。应使用随机抖动，并设置：
- 最大尝试次数；
- 最大总时间；
- 最大退避时间；
- 可重试错误集合；
- 幂等键或状态查询；
- 对
Retry-After的尊重。
十三、几个容易混淆的结论
1. 连接池不是缓存
连接池缓存的是底层连接资源，不是 HTTP 响应。它不会让业务数据自动变成缓存，也不等价于 HTTP 缓存。
连接池：减少 TCP/TLS 建连成本
HTTP 缓存：减少重复获取资源的成本
二者解决的是不同问题。
2. 超时不是取消服务端执行
客户端超时后，通常只能停止等待或关闭本地连接：
客户端：我不等了
服务端：可能仍在继续执行
因此，超时后的写操作仍然需要考虑服务端是否已生效。对于支付、下单、发送消息等操作，应使用幂等键、查询接口或事务状态，而不是仅靠客户端重试。
3. 流式读取不是自动限速
流式 API 只改变数据交付方式，不自动保证：
- 下游处理速度；
- 磁盘写入速度；
- 内存上限；
- 连接及时释放；
- 取消时服务端停止生成数据。
应用仍需要在上下文管理器中消费或关闭响应，并为大文件、慢消费者和取消路径设计资源释放。
4. TLS 加密不等于业务安全
TLS 可以保护客户端与服务端之间的链路，但不能防止：
- 客户端把密码写入日志；
- 服务端返回恶意 JSON；
- API 权限配置错误；
- 重放业务请求；
- 应用层越权；
- 代理或服务端本身泄露数据。
HTTPS 是传输层安全，不会自动替代认证、授权、幂等、审计和输入校验。
结语：把 HTTP 客户端当成一个有状态的资源管理器
一个可靠的 Python HTTP 客户端至少应明确以下关系：
连接池
决定连接如何创建、等待、复用和关闭
超时
决定每个网络阶段最多等待多久
重试
决定哪些失败可以再次发送，以及是否可能重复执行
流式
决定响应或请求体如何分块传输、消费和释放
TLS
决定通信双方如何认证，以及数据如何加密和校验
其中最重要的因果链是：
客户端复用连接
→ 减少 TCP/TLS 握手
→ 降低单次请求成本
客户端限制连接数
→ 并发超过上限时进入池等待
→ 等待过久触发 PoolTimeout
客户端设置读取超时
→ 只能限制单次读取等待
→ 不代表服务端没有执行请求
客户端自动重试
→ 可能再次产生业务副作用
→ 必须结合幂等性和请求可重复性
客户端流式读取
→ 降低单次内存占用
→ 但必须正确关闭响应并处理背压
客户端验证 TLS
→ 确认连接对端身份
→ 不能通过 verify=False 作为生产修复
当这些阶段被分别配置、监控和测试时，HTTP 客户端才不再是一个隐藏网络细节的函数调用，而是一个可以推理、诊断和控制的系统组件。
系列导航与关联阅读
- 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
- 上一篇：Python Socket 编程：TCP、UDP、地址、半关闭、超时和协议帧
- 下一篇：Python 哈希、HMAC 与 TLS：完整性、密码存储、证书和误区
- 延伸：Python Web API 工程：契约、错误、分页、幂等、限流和版本
官方资料
本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。
评论
0 条讨论