一、问题背景：线程池溢出，接口拖垮整个服务
去年接手一个内部订单报表服务，核心接口/api/v1/orders/summary需要聚合订单服务、用户服务、商品服务三个下游系统的数据。最初实现非常简单粗暴——用requests同步调用，代码长这样：
# before.py - 同步串行版本
def get_summary(order_id: str):
order = requests.get(f"http://order-svc/orders/{order_id}", timeout=2).json()
user = requests.get(f"http://user-svc/users/{order['user_id']}", timeout=2).json()
product = requests.get(f"http://product-svc/products/{order['product_id']}", timeout=2).json()
return {"order": order, "user": user, "product": product}
三个请求串行，每个下游平均延迟200ms，接口P95直接到680ms。更糟的是，Gunicorn配置了20个worker，每个worker 10个线程，高峰期每秒50个请求就把线程池塞满，导致其他轻量接口也排队。当时线上告警不断，用户反馈「页面转圈转得怀疑人生」。
二、环境与版本：Python 3.10 + FastAPI + wrk压测
明确一下改造环境，方便你复现对比：
Python 3.10.8
FastAPI 0.95.1 (用于暴露新接口)
Uvicorn 0.21.1 (ASGI服务器，worker数=4)
httpx 0.24.1 (支持异步的HTTP客户端)
asyncio 内置 (Python 3.10原生支持)
wrk 4.2.0 (压测工具)
压测命令统一为：wrk -t4 -c50 -d30s http://localhost:8000/api/v1/orders/summary，即4线程、50并发、持续30秒。
三、方案设计：asyncio + httpx + 信号量限流
核心思路很简单：三个下游调用没有数据依赖，完全可以用asyncio.gather并发执行。但有几个坑必须提前设计：
- 连接复用：每次请求都新建连接很浪费，必须用
httpx.AsyncClient作为全局单例，复用连接池。 - 限流保护：如果下游服务扛不住突发并发，我们会把它打挂。所以用
asyncio.Semaphore限制最大并发数为20。 - 超时与重试：单次超时设为1.5秒（比同步版更短，因为并发能容忍一次重试），重试1次，避免下游抖动导致全链路失败。
- 错误隔离：一个下游挂了不能影响另外两个，用
asyncio.gather(..., return_exceptions=True)捕获异常，返回降级数据。
四、核心实现：完整可运行的异步重构代码
# after.py - asyncio并发版本
import asyncio
import httpx
from fastapi import FastAPI
from contextlib import asynccontextmanager
app = FastAPI()
# 全局共享的AsyncClient，复用连接池
_client: httpx.AsyncClient = None
_semaphore = asyncio.Semaphore(20) # 限制最大并发20
async def fetch_with_limit(client: httpx.AsyncClient, url: str, timeout: float = 1.5):
"""带信号量限流和单次重试的请求函数"""
async with _semaphore:
try:
resp = await client.get(url, timeout=timeout)
resp.raise_for_status()
return resp.json()
except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
# 重试一次，避免瞬时故障
try:
resp = await client.get(url, timeout=timeout)
resp.raise_for_status()
return resp.json()
except Exception as retry_e:
return {"error": str(retry_e), "fallback": True}
@asynccontextmanager
async def lifespan(app: FastAPI):
global _client
# 连接池参数：单host最大10个连接，空闲keep-alive 30秒
_client = httpx.AsyncClient(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
timeout=httpx.Timeout(connect=1.0, read=1.5, write=1.0, pool=1.0)
)
yield
await _client.aclose()
app.router.lifespan_context = lifespan
@app.get("/api/v1/orders/summary")
async def get_summary(order_id: str):
# 三个请求并发执行，互不阻塞
order_task = fetch_with_limit(_client, f"http://order-svc/orders/{order_id}")
user_task = fetch_with_limit(_client, f"http://user-svc/users/{order_id}") # 简化演示
product_task = fetch_with_limit(_client, f"http://product-svc/products/{order_id}")
results = await asyncio.gather(order_task, user_task, product_task, return_exceptions=True)
# 解析结果，出错时返回降级信息
order, user, product = results
return {
"order": order if not isinstance(order, dict) or "error" not in order else {},
"user": user if not isinstance(user, dict) or "error" not in user else {},
"product": product if not isinstance(product, dict) or "error" not in product else {},
"degraded": any(isinstance(r, dict) and r.get("error") for r in results)
}
关键点说明：
- httpx.AsyncClient必须在lifespan中初始化，避免每次请求创建新连接（那是灾难）。
- Semaphore(20)意味着即使50并发打过来，对下游的并发请求峰值也只有20，不会压垮依赖服务。
- asyncio.gather的return_exceptions=True保证即使某个任务抛出未捕获异常，其他任务结果也能正常返回。
五、踩坑与优化：三个必须注意的细节
1. 事件循环阻塞陷阱
一开始我用requests库配合asyncio.to_thread，发现性能提升有限。原因：asyncio.to_thread本质是丢到线程池，依然受GIL和线程切换开销影响。换成httpx.AsyncClient后，IO操作真正走epoll，CPU占用从40%降到15%。记住：异步编程必须用异步IO库，否则白搭。
2. 连接池参数调优
默认httpx连接池是10个keepalive连接，压测时发现Too many open connections错误。调大max_keepalive_connections=20后稳定。另外必须设置pool超时，否则连接池取不到连接时会无限等待，表现为接口卡死。
3. 信号量位置
信号量放在fetch_with_limit内部而不是外部，这样每个请求独立限流。如果放在gather外面，所有并发请求共享一个信号量，会导致部分请求排队过久，P99不降反升。实测放内部比外部P99低40%。
六、效果数据：wrk压测对比
同一套逻辑，同一台机器（4核8G），用wrk压测30秒，数据如下：
| 指标 | 同步版本 (before) | 异步版本 (after) | 提升幅度 |
|---|---|---|---|
| 平均延迟 | 680ms | 120ms | 5.7倍 |
| P50 | 650ms | 105ms | 6.2倍 |
| P95 | 680ms | 180ms | 3.8倍 |
| P99 | 1.2s | 300ms | 4倍 |
| 吞吐量 (RPS) | 72 | 415 | 5.8倍 |
| CPU占用 | 40% | 15% | 下降62% |
压测输出片段（异步版）：
Running 30s test @ http://localhost:8000/api/v1/orders/summary
4 threads and 50 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 120.25ms 38.92ms 380.11ms 82.13%
Req/Sec 104.23 18.55 148.00 67.90%
12450 requests in 30.00s, 2.02MB read
Requests/sec: 415.03
Transfer/sec: 68.92KB
七、总结与建议
这次重构的核心收益不是代码量减少，而是将串行等待转化为并发等待，同时通过信号量保护了下游系统。几点实战建议：
- 不是所有场景都适合asyncio：如果是CPU密集型任务，异步没用，用
multiprocessing更合适。IO密集型（HTTP调用、数据库查询、文件读写）才是asyncio的主场。 - 一定要配合连接池：
httpx.AsyncClient是异步版的requests.Session，复用连接能省掉TCP握手和TLS协商的20-30ms。 - 监控P99而非平均延迟：异步化后平均延迟好看，但P99更能反映限流和超时配置是否合理。
- 渐进式改造：如果老项目是Flask/Django同步框架，别一次性重写。可以先用
asyncio.run在视图函数里跑异步代码，或者用gunicorn+uvicorn worker平滑过渡。
如果你也遇到类似接口延迟问题，建议先画一下依赖关系图，找出无依赖的串行调用，用asyncio并发化——这通常是性价比最高的优化手段。有问题欢迎评论区交流，我会尽力解答。
