必须用async with httpx.asyncclient()包裹，否则连接池失效、触发“too many open files”错误，并可能抛出runtimewarning；正确做法是全程复用同一client实例并配合semaphore限流。
async with httpx.AsyncClient() 是必须的起点
不显式创建和关闭异步客户端，httpx 会在每次请求时新建连接，导致连接池失效、TIME_WAIT 暴增，微服务批量调用时容易触发“Too many open files”错误。Python 3.11 的 asyncio 对未关闭的异步上下文管理器更敏感，可能抛出 RuntimeWarning: coroutine 'AsyncClient.__aexit__' was never awaited。
正确做法是始终用 async with 包裹整个批处理过程：
async def batch_call_services(urls):
async with httpx.AsyncClient(timeout=5.0) as client:
tasks = [client.get(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
- 不要在循环里反复
AsyncClient()—— 即使加了await client.aclose()也不如async with可靠 -
timeout建议显式设为浮点数（如5.0），Python 3.11 中传整数可能被误判为连接超时而非读取超时 - 若需复用同一 client 处理多个批次，应将
AsyncClient实例作为参数传入或封装为类属性，而非每次重建
并发数控制不当会压垮下游微服务
直接 asyncio.gather(*[client.get(u) for u in urls]) 等同于无限制并发，几十个 URL 同时发出去，下游服务（尤其是 Java Spring Boot 默认 Tomcat 线程池仅 200）大概率 503 或响应延迟陡增。
必须用 asyncio.Semaphore 限流：
sem = asyncio.Semaphore(10) # 全局并发上限设为 10 <p>async def fetch_with_limit(client, url): async with sem: return await client.get(url)</p><p>async def batch_call_services(urls): async with httpx.AsyncClient() as client: tasks = [fetch_with_limit(client, url) for url in urls] return await asyncio.gather(*tasks, return_exceptions=True) </p>
- 把
Semaphore放在async with httpx.AsyncClient()外层，避免每次新建 client 重置信号量 - 阈值不是拍脑袋：建议从 5 开始压测，观察下游服务的 CPU、GC 和 HTTP 5xx 比例再逐步上调
- 别用
asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)手动调度——复杂且易漏异常，gather+return_exceptions=True更稳
JSON 解析失败常因响应非 2xx 但没检查 status_code
微服务间调用常见 400/404/500 响应体仍含 JSON（比如 {"error": "not found"}），但直接 response.json() 会抛 httpx.HTTPStatusError，导致整个 gather 中断，后续请求结果丢失。
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
必须在拿到 response 后先判断状态，再解析：
async def safe_fetch(client, url):
try:
resp = await client.get(url)
if resp.is_success: # 即 status_code in [200, 299]
return {"url": url, "data": resp.json(), "error": None}
else:
return {"url": url, "data": None, "error": f"HTTP {resp.status_code}"}
except httpx.HTTPError as e:
return {"url": url, "data": None, "error": str(e)}
- 别依赖
raise_for_status()—— 它默认抛异常，违背“批量容错”目标 - 微服务返回的 JSON 字段名不统一很常见，
resp.json()前最好加try/except json.JSONDecodeError，有些服务出错时返回 HTML 或空体 - Python 3.11 中
httpx默认启用http2=True，若下游服务不支持 HTTP/2，会静默降级到 HTTP/1.1，但某些网关（如早期 Envoy）可能返回畸形响应，此时resp.text可能是乱码，优先看resp.content和resp.headers.get("content-type")
Cookie、Header 和 auth 在批量调用中容易被全局污染
同一个 AsyncClient 实例复用时，若某次请求设置了 cookies 或 headers，它们会持续影响后续请求——微服务 A 需要 X-Trace-ID，微服务 B 要求禁止该头，不隔离就会出问题。
解决方案只有两个：要么按服务分 client，要么每次请求显式覆盖：
# 方案一：按服务类型建不同 client（推荐）
svc_a_client = httpx.AsyncClient(headers={"X-Service": "A"})
svc_b_client = httpx.AsyncClient(headers={"X-Service": "B"})
<h1>方案二：单 client + 每次传参（适合 header 差异小）</h1><p>await client.get(url, headers={"X-Custom": "value", **common_headers})
</p>
-
AsyncClient的cookies参数是 session 级的，无法 per-request 覆盖，必须用方案一 - Basic Auth 可以用
auth=("user", "pass")per-request 传，但 Token 类认证（如 Bearer）建议走headers，否则 token 过期后所有请求都失败 - Python 3.11 的
httpx1.x 版本对default_encoding处理更严格，若微服务返回的Content-Type缺少charset，response.text可能解码失败，此时应改用response.content.decode("utf-8", errors="replace")
真正麻烦的从来不是并发本身，而是每个服务对超时、重试、认证、编码的隐式假设——批量调用前，先用 curl -v 抓几组真实响应头和 body，比读文档管用。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
