Python httpx 怎样复用异步客户端连接

我第一次把 HTTPX 接进异步任务时，最容易写出的代码是“每次调用都新建一个 AsyncClient”。功能当然能跑，但请求一多，客户端的生命周期就和业务循环绑死了，连接池也很难真正复用。

更稳妥的做法是：让一个 httpx.AsyncClient 覆盖一段明确的异步生命周期，把它传给需要发请求的函数；任务结束时统一关闭。HTTPX 官方入口是 https://www.python-httpx.org/，客户端文档明确说明，客户端实例会使用连接池，异步客户端还可以被多个任务共享。

如果不是一次性实验，就不要在热循环里反复创建 AsyncClient。先确定客户端作用域，再用 Limits 控制连接池容量，最后把关闭动作放在这个作用域的出口。

问题：为什么循环里新建 AsyncClient 会失去复用收益

顶层的 httpx.get() 适合少量测试请求；需要连续访问同一服务时，客户端实例才会维护连接池。常见误区是把 async with httpx.AsyncClient() 放进热循环：每轮都会新建并关闭一个池，即使代码用了异步，也没有把多个请求放进同一个连接资源边界。

# 反例：每次循环都创建一个新的客户端，连接池无法跨轮复用
async def load_items(paths: list[str]) -> list[int]:
    statuses = []
    for path in paths:
        async with httpx.AsyncClient() as client:
            # 这一轮结束后客户端关闭，下一轮不会接着使用同一连接池
            response = await client.get(path)
            statuses.append(response.status_code)
    return statuses

这里的问题不是 async with 本身，而是作用域太短。一次请求一个客户端会增加连接建立、握手和资源回收的次数；它还让公共请求头、超时和连接池参数散落在调用点。

最小配方：用 async with 管理一个共享客户端

先把客户端提升到覆盖整组请求的作用域，再把它作为参数传入。这样连接池属于这组任务，而不是属于某一个路径。

import httpx

async def load_items(paths: list[str]) -> list[int]:
    # 一个客户端覆盖整组请求，退出 async with 时统一释放连接资源
    async with httpx.AsyncClient(timeout=10.0) as client:
        async def fetch(path: str) -> int:
            # 多个任务共享同一个客户端，连接池可以复用已有连接
            response = await client.get(path)
            response.raise_for_status()
            return response.status_code

        # gather 只组织并发任务，不额外创建客户端实例
        return await asyncio.gather(*(fetch(path) for path in paths))

示例里的 asyncio.gather 需要在文件顶部导入 asyncio。客户端文档推荐用上下文管理器，手动管理时则要在确定的清理路径调用 await client.aclose()。

关键参数：用 Limits 约束连接池容量

复用不是“连接越多越好”。HTTPX 的 Limits 允许你分别设置最大连接数、最大保活连接数和空闲保活连接的存活时间。对我来说，先把这三个参数和上游服务的并发承受能力对齐，再决定是否打开 HTTP/2，会比盲目提高并发更可控。

limits = httpx.Limits(
    max_connections=10,          # 限制同时占用的连接总数
    max_keepalive_connections=5, # 限制池中保留的空闲连接数
    keepalive_expiry=5.0,        # 空闲连接超过秒数后允许回收
)

# 把连接池策略放在客户端边界，调用函数只关心请求本身
client = httpx.AsyncClient(limits=limits, timeout=10.0)

官方文档给出的默认值是 max_connections=100、max_keepalive_connections=20、keepalive_expiry=5。这些是库的默认边界，不是任何业务服务的推荐值；连接数仍应结合上游限流、任务并发和响应耗时调整。

变体与坑点：服务级共享、并发任务和流式响应

如果调用发生在长生命周期的异步服务里，可以让应用启动时创建一个客户端，在应用关闭时调用 aclose()。如果只是一个批处理函数，函数内的 async with 已经足够。关键判断是：客户端要覆盖一组相关请求，但不能跨越你无法管理的事件循环或关闭边界。

HTTPX API 文档说明 AsyncClient 可以被多个任务共享，因此可以把同一个实例传给并发任务；不要在每个任务内部再创建一个客户端。另一个容易忽略的边界是手动流式响应：如果使用 client.send(..., stream=True)，需要最终调用 await response.aclose()，否则响应占用的连接可能长期留在池里。

• 一次性脚本：用 async with AsyncClient() 包住整段请求。
• 异步服务：把客户端放进服务生命周期，关闭时调用 aclose()。
• 高并发批处理：共享客户端，再用 Limits 给连接总数设上限。

完整片段：一个可复用的异步请求封装

下面的写法把连接池配置、公共请求头和清理时机集中在一个工厂函数里；业务函数只接收客户端和路径。它适合继续扩展重试、鉴权或统一错误记录，但这些策略不应通过重复创建客户端来实现。

import asyncio
from collections.abc import Sequence

import httpx

async def fetch_statuses(paths: Sequence[str]) -> list[int]:
    # 用一个池承载本次批量请求，并限制并发连接规模
    limits = httpx.Limits(max_connections=10, max_keepalive_connections=5)
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"User-Agent": "sample-httpx-client/1.0"},  # 集中设置公共请求头
        timeout=10.0,
        limits=limits,
    ) as client:
        async def fetch(path: str) -> int:
            # 每个任务复用外层客户端，异常会沿调用链抛出
            response = await client.get(path)
            response.raise_for_status()
            return response.status_code

        # gather 共享同一个连接池，作用域退出时客户端自动关闭
        return list(await asyncio.gather(*(fetch(path) for path in paths)))

这段代码的取舍很明确：客户端作用域覆盖一批请求，连接池参数集中管理，并发任务共享客户端，异常不被静默吞掉。若改成手动创建客户端，就必须在 finally 中调用 await client.aclose()；若使用流式模式，还要单独保证响应关闭。

常见问题

AsyncClient 应该每个请求创建一次吗？

通常不应该。让一个客户端覆盖合理的请求组，才能复用连接池；只有完全独立的一次性调用才适合短作用域客户端。

共享客户端会不会让并发失控？

共享本身不是无限并发，真正的连接上限由 Limits 和外层任务调度共同决定。先设置资源上限，再根据服务端限制调整。

什么时候必须手动关闭？

不使用 async with 时必须调用 await client.aclose()；手动流式响应也要保证 await response.aclose() 最终执行。

我最后会把这条规则记成一句话：客户端是连接资源的边界，不是请求函数里的临时变量。先确定生命周期，再共享 AsyncClient，最后用 Limits 把并发和空闲连接控制在可解释的范围内。