投稿
  1. 其其网首页
  2. 投稿

httpx 异步 HTTP 客户端使用方法（AI 视频下载总结器的 HTTP 工具）

其AI • • 投稿

下载视频、抓缩略图、调 DeepSeek、对接 Stripe、转写音频，每个环节都离不开 HTTP 调用。一开始有人用 requests，同步调用把事件循环卡死；有人用 aiohttp，写法又绕，会话管理、超时配置都比预想繁琐。我最后统一用 httpx 做异步 HTTP 客户端，一套 API 同时支持同步和异步，迁移成本还低。下面是我在这个项目里积累的完整实践，从选型到踩坑再到性能优化。

一、为什么选 httpx

选型的时候我把主流方案摆在一起比过一轮，结论很干脆。

库 同步/异步 类型提示 性能 维护
requests 同步 弱 中 活跃
aiohttp 异步 弱 高 活跃
httpx 同步+异步 ✅ 高 活跃

平台选 httpx 因为：

  • 同步+异步统一 API；
  • 类型提示好；
  • HTTP/2 支持；
  • 与 requests 兼容（迁移成本低）。

requests 的同步阻塞不适合异步项目，一旦混进 async 函数，整个事件循环都要等它；aiohttp 功能确实全，但学习成本摆在那里。httpx 的同步接口和 requests 几乎一致，团队迁移只需要把 requests.get 换成 httpx.get，异步场景再换 AsyncClient，这是它胜出的关键。

二、基础用法

异步客户端的三个基础场景：GET、POST、流式读取。日常开发里八成以上的调用就这三种，先把它们写熟，其他高级特性都是在这上面叠加的。

import httpx

# 异步 GET
async with httpx.AsyncClient() as client:
    response = await client.get('https://api.example.com/data')
    data = response.json()

# 异步 POST
async with httpx.AsyncClient() as client:
    response = await client.post(
        'https://api.example.com/submit',
        json={'key': 'value'}
    )

# 流式
async with httpx.AsyncClient() as client:
    async with client.stream('GET', url) as response:
        async for chunk in response.aiter_bytes():
            process(chunk)

注意这里用 async with 管理客户端生命周期，退出时自动关闭连接，不会泄漏文件描述符。流式场景用 aiter_bytes 按块读取，响应再大也不会整块塞进内存，这是后面大文件下载能顺利落地的根基。

三、全局 Client 配置

每个请求都 new 一个 client 是浪费，连接没法复用，TLS 握手反复进行，高并发下文件描述符也顶不住。我把它放进 FastAPI 的 lifespan，应用启动时创建，退出时统一关闭。

# main.py
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动
    app.state.http = httpx.AsyncClient(
        timeout=30.0,
        follow_redirects=True,
        limits=httpx.Limits(
            max_connections=100,
            max_keepalive_connections=20
        ),
        headers={'User-Agent': 'AI-Video-Summarizer/1.0'}
    )
    yield
    # 关闭
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

依赖注入：

async def get_http(request: Request) -> httpx.AsyncClient:
    return request.app.state.http

@router.get('/api/proxy')
async def proxy(http: httpx.AsyncClient = Depends(get_http)):
    response = await http.get('https://api.example.com')
    return response.json()

全局 client 带统一超时、统一 User-Agent、连接池上限，所有接口共享。通过 Depends 注入后，每个接口拿到的是同一个实例，连接被充分复用，握手次数大幅下降。lifespan 里的 aclose 保证进程退出前连接优雅关闭，不会留下半开连接。

四、超时控制

不同场景对超时的容忍完全不同：普通 API 秒级就该超时，下载视频要几分钟。如果全用同一个超时值，要么普通接口被慢请求拖死，要么大文件下载中途断掉，所以我把超时按场景拆成几档。

class TimeoutConfig:
    FAST = 5.0       # 普通 API
    NORMAL = 30.0    # 视频信息
    SLOW = 120.0     # 下载文件
    STREAM = None    # 流式不限时

# 用
client = httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=5.0,
        read=30.0,
        write=10.0,
        pool=5.0
    )
)
# 单次超时
response = await client.get(url, timeout=10.0)

connect、read、write、pool 分别设置，比单一超时精准：连接 5 秒建不起来就该放弃，读 30 秒没数据说明对端有问题。个别接口还能用单次 timeout 覆盖默认值，比如健康检查用 5 秒快速失败，慢的节点直接被摘掉。

五、限流

下载接口对外部服务有配额，并发一高就被对方限流甚至封 IP，全链路都会跟着遭殃。我在应用层用信号量控制并发，先把自家请求控制在配额内。

import asyncio

semaphore = asyncio.Semaphore(10)  # 最多 10 并发

async def fetch_with_limit(client, url):
    async with semaphore:
        return await client.get(url)

信号量把并发钉死在 10，再多请求也在门口排队。这个值要按对端配额和单请求耗时来调：太小浪费带宽，吞吐上不去；太大会触发对端限流，得不偿失。上线后看实际耗时不限流，再微调数字。

六、文件下载

视频文件从几十 MB 到几 GB，下载方式必须分场景，不能一把梭。

1. 完整下载到内存

async def download_file(client, url: str) -> bytes:
    response = await client.get(url)
    response.raise_for_status()
    return response.content

2. 流式下载到磁盘

async def download_to_file(client, url: str, path: str):
    async with client.stream('GET', url, follow_redirects=True) as response:
        response.raise_for_status()
        
        with open(path, 'wb') as f:
            async for chunk in response.aiter_bytes(chunk_size=8192):
                if chunk:
                    f.write(chunk)

平台用这个下载视频文件（几十 MB 到几 GB）。

3. 进度回调

async def download_with_progress(client, url: str, path: str, 
                                   on_progress=None):
    async with client.stream('GET', url) as response:
        total = int(response.headers.get('content-length', 0))
        downloaded = 0
        
        with open(path, 'wb') as f:
            async for chunk in response.aiter_bytes(8192):
                f.write(chunk)
                downloaded += len(chunk)
                if on_progress and total:
                    on_progress(downloaded / total)

小文件直接 content 一把梭，几 MB 无压力；大文件必须走流式，内存占用稳定在几十 MB。进度回调基于 content-length 计算百分比，前端能实时看到下载进度条；对端不返回 content-length 时 total 为 0，回调自动跳过，不会出现除零错误。

七、SSE 流式响应

DeepSeek 的总结是流式返回的，用户要看到逐字生成的效果，必须解析 SSE 事件流，而不是傻等整个响应。

async def stream_deepseek(prompt: str):
    async with httpx.AsyncClient() as client:
        async with client.stream(
            'POST',
            'https://api.deepseek.com/v1/chat/completions',
            headers={'Authorization': f'Bearer {API_KEY}'},
            json={
                'model': 'deepseek-chat',
                'messages': [{'role': 'user', 'content': prompt}],
                'stream': True
            }
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith('data: '):
                    data = line[6:]
                    if data == '[DONE]':
                        break
                    chunk = json.loads(data)
                    delta = chunk['choices'][0]['delta'].get('content', '')
                    if delta:
                        yield delta

这里用 aiter_lines 按行读，识别 data: 前缀，逐个 yield 增量内容，SSE 的跨行分块由 httpx 内部处理好。一个容易忽略的点：如果你自己攒 buffer 做行拼接，要小心单行被 TCP 分包切碎的情况，按行读则没这个烦恼。

八、并发请求

批量总结多个视频时，一个个串行请求太慢，用户等不起。我用 asyncio.gather 并发跑，把多个请求的耗时压到单次请求的水平。

async def fetch_multiple(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        return [
            r.json() if not isinstance(r, Exception) else {'error': str(r)}
            for r in responses
        ]

return_exceptions=True 很关键，某个请求失败不会拖垮整个 gather，错误被收进结果列表单独标记。这里要配合信号量使用，否则并发数不可控，容易重演”没限制并发”那个坑。

九、重试机制

httpx 不内置重试，遇到网络抖动或 5xx，请求就直接失败。我用 tenacity 补上指数退避重试，让瞬时故障自己恢复。

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_with_retry(client, url):
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

针对特定状态码重试：

from httpx import HTTPStatusError

@retry(
    stop=stop_after_attempt(3),
    retry=retry_if_exception_type(HTTPStatusError),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_api(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        if response.status_code in (429, 500, 502, 503, 504):
            response.raise_for_status()  # 触发重试
        return response.json()

重试只针对可恢复的错误：429 限流、5xx 服务端故障、网络异常。4xx 业务错误（参数不对、没权限）重试也没用，别浪费请求次数。指数退避避免重试风暴，多实例同时失败时不会在同一秒全部打向对端。

十、HTTP/2 支持

和同一域名的大量并发请求，HTTP/2 的多路复用能把性能拉起来，一个连接顶一堆连接。

client = httpx.AsyncClient(http2=True)

需要安装 h2：

pip install httpx[http2]

HTTP/2 优势：

  • 多路复用（一个连接并发多个请求）；
  • Header 压缩；
  • 服务器推送。

开启后同一个 TCP 连接上可以并发多个请求，省掉大量握手开销。但部分老后端只支持 HTTP/1.1，开启后连接会失败，需要先确认目标服务支持，或者做降级判断，别一开就全部请求挂掉。

十一、Cookie 管理

有些场景需要保持会话，比如带登录态的请求，Cookie 管理就要接进来。

client = httpx.AsyncClient(
    cookies={'session': 'abc123'}
)

# 也可以在请求时设
response = await client.get(url, cookies={'extra': 'value'})

Cookie Jar：

jar = httpx.Cookies()
jar.set('session', 'abc123', domain='example.com')

client = httpx.AsyncClient(cookies=jar)

客户端创建时传 cookies，请求期间 Cookie Jar 会自动管理 Set-Cookie，响应里设置的新 Cookie 也会被记住。注意 SameSite 策略，跨站请求时 Cookie 不会自动带上，这个坑我们踩过，后来靠显式指定域和 path 解决。

十二、代理

项目里有部分接口必须走代理才能访问，代理配置也走 client 层统一处理。

# HTTP 代理
client = httpx.AsyncClient(
    proxy='http://proxy.example.com:8080'
)

# 多个代理随机
import random
PROXIES = ['http://p1:8080', 'http://p2:8080']
client = httpx.AsyncClient(proxy=random.choice(PROXIES))

单代理配置简单，适合固定出口的场景；多代理随机分摊流量，适合代理节点经常失效的情况。多代理要盯着节点健康，失效的代理会让请求连环超时，宁可少几个节点也要保证存活。

十三、SSL 配置

内部服务用了自签名证书，默认校验会直接失败，需要指定 CA 文件。

# 自定义 CA
client = httpx.AsyncClient(
    verify='/path/to/ca-bundle.crt'
)

# 跳过验证（不推荐）
client = httpx.AsyncClient(verify=False)

verify 指向自定义 CA 文件，内网服务就能正常访问。跳过验证只适合本地调试，生产环境禁用，否则中间人攻击毫无防护，这是安全红线，团队规范里写得很清楚。

十四、文件上传

把视频上传到存储服务，用 multipart 表单，文件和相关字段一起提交。

async def upload_file(client, url: str, file_path: str):
    with open(file_path, 'rb') as f:
        files = {'file': (file_path, f, 'application/octet-stream')}
        response = await client.post(url, files=files)
    return response.json()

multipart/form-data：

files = {
    'file': (
        'video.mp4',           # filename
        open('video.mp4', 'rb'),
        'video/mp4'            # content-type
    )
}
data = {'description': 'my video'}
response = await client.post(url, files=files, data=data)

files 参数会自动编码成 multipart/form-data，data 里的字段和文件一起提交。文件对象要在 with 块里打开，保证读取期间文件句柄不关闭；如果没带 content-type，服务端可能猜错文件类型，所以显式写全。

十五、Webhook 验证

Stripe 的支付回调必须验签，否则伪造请求就能篡改订单状态。验签放接口入口，第一时间拦截。

async def verify_stripe_webhook(client, payload: bytes, 
                                  signature: str) -> dict:
    async with client as c:
        event = stripe.Webhook.construct_event(
            payload, signature, WEBHOOK_SECRET
        )
    return event

验签用 Stripe 官方 SDK，本地校验签名，验不过直接抛异常。Webhook 接口要尽快返回 200，处理逻辑丢到后台任务，避免对端超时重推造成重复处理，幂等校验也要在业务侧补上。

十六、错误处理

外部调用必须分类处理异常，不能一把梭 try-except 把错误全吞掉，否则线上问题全靠猜。

from httpx import (
    HTTPStatusError, RequestError, 
    TimeoutException, ConnectError
)

try:
    response = await client.get(url, timeout=10.0)
    response.raise_for_status()
except TimeoutException:
    log.error('请求超时')
except ConnectError:
    log.error('连接失败')
except HTTPStatusError as e:
    log.error(f'HTTP 错误: {e.response.status_code}')
    if e.response.status_code == 429:
        retry_after = e.response.headers.get('Retry-After')
        log.warning(f'限流，{retry_after}秒后重试')
except RequestError as e:
    log.error(f'请求错误: {e}')

先捕具体的，再捕兜底的 RequestError，顺序别写反。429 时读取 Retry-After 头按服务端要求等待，这是对限流服务的基本尊重。错误日志里带上状态码和耗时，排查问题时不用翻原始请求记录。

十七、连接池

连接池配置直接决定并发能力和资源占用，是全局 client 里影响面较大的参数。

client = httpx.AsyncClient(
    limits=httpx.Limits(
        max_connections=100,         # 总连接数
        max_keepalive_connections=20,  # keep-alive 数
        keepalive_expiry=30          # keep-alive 时间
    )
)

复用连接减少握手开销。

maxconnections 是总连接数上限，maxkeepaliveconnections 是保留的空闲连接数，keepaliveexpiry 控制空闲连接存活时间。这几个参数按业务并发量调：太小会排队等待，吞吐上不去；太大会占满文件描述符，进程反而报错。上线后用监控数据反推，别拍脑袋。

十八、踩过的坑

真实项目里踩过的坑按严重程度列出来，每一个都对应过一次线上问题。

  • 没限制并发：1000 个并发请求把服务端打挂。Semaphore 限流。
  • 同步调用阻塞事件循环：requests 库会阻塞。必须用 httpx async。
  • 未关 Client：每次请求都新建 client 浪费资源。复用全局 client。
  • 超时未设：httpx 默认 5s 不够。设 30s。
  • 文件下载 OOM：1GB 视频加载到内存爆。流式下载。
  • SSE 解析：单行可能跨 chunk 边界。攒 buffer。
  • HTTP/2 兼容性：部分老后端不支持，连接失败。
  • Cookie SameSite：跨站请求 Cookie 不带。
  • DNS 解析慢：连接前 DNS 查询可能 1s+。预解析或 usednscache。
  • 异步上下文管理：async with 退出时未关闭。手动 aclose()。

这里最伤的是没限制并发和同步调用阻塞：前者把对端打挂，后者把自己卡死，都直接引发线上事故。OOM 问题在下载 1GB 视频时真实发生过，改成流式后内存稳定在几十 MB。DNS 解析慢属于隐藏问题，高并发时会被放大，值得提前处理。

十九、性能优化

性能优化的几个抓手按性价比排序，先做收益明显的。

# 1. 复用连接
client = httpx.AsyncClient()  # 全局一个

# 2. HTTP/2 多路复用
client = httpx.AsyncClient(http2=True)

# 3. DNS 缓存
client = httpx.AsyncClient(
    transport=httpx.AsyncHTTPTransport(retries=2)
)

# 4. 限制超时
client = httpx.AsyncClient(timeout=10.0)

复用连接收益明显且几乎零成本，先做；HTTP/2 对同域并发帮助大，次之；超时限制避免慢请求拖死连接池，属于防御性优化。DNS 缓存和 transport 重试放在最后，按实际瓶颈决定要不要上。

二十、与第三方库对比

同样一个请求，三个库的写法差异很大，放在一起看最直观。

# requests (同步)
import requests
response = requests.get(url)  # 阻塞

# aiohttp (异步，但 API 复杂)
import aiohttp
async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        data = await response.json()

# httpx (统一 API)
import httpx
async with httpx.AsyncClient() as client:
    response = await client.get(url)  # 与 requests 几乎一样
    data = response.json()

requests 一行搞定但同步阻塞；aiohttp 异步但会话管理、超时配置都更繁琐；httpx 的写法最接近 requests，心智负担低。这也是我把它定为全项目统一 HTTP 工具的原因，新人接手代码不需要重新学一套 API。

二十一、最佳实践清单

收敛成八条写进项目开发规范，作为 HTTP 调用相关代码的审查标准。

  1. 全局 client 复用；
  2. 不同场景不同超时；
  3. 流式下载大文件；
  4. 重试用 tenacity；
  5. 并发用 Semaphore；
  6. 错误分类处理；
  7. HTTP/2 多路复用；
  8. 资源自动释放（lifespan）。

到这里，httpx 从选型到性能优化的实践链路就完整了。照这套走，外部接口调用基本不会再有资源泄漏、阻塞和并发失控的问题，后续接入新的第三方服务也只需要照着模板加一层封装。

常见问题（FAQ）

Q1：httpx 替代 requests？

替代。同步 API 兼容，迁移成本低。

Q2：httpx 怎么用 HTTP/2？

httpx[http2] 装依赖，AsyncClient(http2=True) 开启。

Q3：流式响应怎么解析 SSE？

aiter_lines() 按行读，识别 data: 前缀。

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 qiqicto@qq.com 举报，一经查实，本站将立刻删除。
赞 (0)
其AI的头像其AI普通用户
0

相关推荐

返回顶部