一、问题背景：一个被同步IO拖死的接口
去年接手了一个用户画像查询服务，逻辑不复杂：接收user_id列表，并发调用3个下游服务（基础属性、行为标签、风控评分），聚合后返回。下游都是HTTP接口，平均RT 80ms左右。
原来的实现是Flask + Gunicorn（sync worker），配置是 --workers=8 --threads=4。问题很明显：
- 单机QPS峰值只有120，高峰期CPU利用率不到15%，但请求全堵在IO等待上
- P99延迟1.8s，用户端频繁超时
- 每次扩容只能加机器，成本线性增长
根因就一个：同步阻塞IO。每个请求要串行等3次HTTP调用，8个worker×4线程=32个并发槽位，槽位一满就排队。CPU闲着，IO在等，典型的资源浪费。
二、环境与版本
改造前后的技术栈：
# Before
Python 3.8.10
Flask 2.0.3
Gunicorn 20.1.0 (sync worker)
requests 2.25.1
# After
Python 3.11.6
FastAPI 0.109.0
uvicorn 0.27.0 (uvloop 0.19.0)
httpx 0.26.0
Python 3.11对asyncio有原生优化（比如TaskGroup、异常组），uvloop在Linux下比默认事件循环快2-4倍，这两个是必选的。
三、方案设计
核心思路三条：
- 把IO等待时间交还给事件循环：3个下游调用从串行改成
asyncio.gather并发，理论耗时就变成max(80ms)而不是sum(240ms) - 连接池复用：用
httpx.AsyncClient全局单例，复用TCP连接，省掉每次TLS握手 - 超时+降级：每个下游设置独立超时（300ms），用
asyncio.wait_for包裹，超时返回默认值，避免单个下游拖垮整体
架构上从「每请求一线程」变成「单进程事件循环 + 多worker」，worker数按CPU核数配置（4核跑4个worker），每个worker内部能扛上千并发。
四、核心实现
Before：同步版本
# app_sync.py
import requests
from flask import Flask, request, jsonify
app = Flask(__name__)
SESSION = requests.Session()
def fetch(url, payload):
try:
resp = SESSION.post(url, json=payload, timeout=0.5)
return resp.json()
except Exception:
return {}
@app.route("/profile", methods=["POST"])
def profile():
user_ids = request.json["user_ids"]
result = []
for uid in user_ids:
# 三个下游串行调用，每个80ms，一个用户就是240ms
base = fetch("http://base-svc/query", {"uid": uid})
behavior = fetch("http://behavior-svc/query", {"uid": uid})
risk = fetch("http://risk-svc/query", {"uid": uid})
result.append({**base, **behavior, "risk": risk})
return jsonify(result)
这段代码的致命伤：for 循环里串行调3次HTTP，一个user_id就要240ms。10个user_id就是2.4秒。
After：异步版本
# app_async.py
import asyncio
import httpx
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# 全局单例客户端，连接池上限200，keep-alive 30s
client = httpx.AsyncClient(
timeout=httpx.Timeout(0.3, connect=0.1),
limits=httpx.Limits(max_connections=200, max_keepalive_connections=100),
http2=True,
)
class ProfileReq(BaseModel):
user_ids: list[str]
async def safe_fetch(url: str, payload: dict, default=None):
"""带超时和降级的下游调用"""
try:
resp = await asyncio.wait_for(
client.post(url, json=payload), timeout=0.3
)
return resp.json()
except (asyncio.TimeoutError, httpx.HTTPError):
return default or {}
async def fetch_one(uid: str) -> dict:
# 三个下游并发，总耗时 = max(各下游RT)，不是sum
base, behavior, risk = await asyncio.gather(
safe_fetch("http://base-svc/query", {"uid": uid}),
safe_fetch("http://behavior-svc/query", {"uid": uid}),
safe_fetch("http://risk-svc/query", {"uid": uid}, default={"score": 0}),
)
return {**base, **behavior, "risk": risk}
@app.post("/profile")
async def profile(req: ProfileReq):
# 所有user_id也并发处理
results = await asyncio.gather(*[fetch_one(uid) for uid in req.user_ids])
return {"data": list(results)}
关键改动点：
asyncio.gather两层并发：用户之间并发、单用户的下游之间也并发safe_fetch用wait_for强制超时，下游挂掉不影响主流程httpx.AsyncClient全局复用，连接池限制200防止打爆下游
启动命令：
uvicorn app_async:app --host 0.0.0.0 --port 8000 \
--workers 4 --loop uvloop --http httptools \
--backlog 2048 --limit-concurrency 2000
--workers 4 对应4核，--limit-concurrency 2000 是单worker最大并发连接数。
五、踩坑与优化
改造过程踩了三个坑，值得记录：
坑1：AsyncClient 每次请求新建，反而更慢
一开始我在每个请求里 async with httpx.AsyncClient() as c，结果QPS只有800。原因是每次都要新建连接池、重新TLS握手。改成全局单例后QPS直接翻倍到1800。AsyncClient必须复用，这是铁律。
坑2：CPU密集操作阻塞事件循环
画像聚合里有一段JSON合并和字段映射，用了 json.loads 大对象，单次耗时40ms左右。这段是同步的，会把整个事件循环卡住，导致其他请求全部延迟。解决方案是丢到 run_in_executor：
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(None, heavy_transform, raw_data)
或者用 asyncio.to_thread（3.9+）更简洁。改完后P99从220ms降到95ms。
坑3：下游连接数爆炸
并发上去后，下游服务被打爆了，报了一堆连接拒绝。因为4个worker×200连接=800个连接，下游扛不住。最后把 max_connections 降到50，配合 --limit-concurrency 1000，在下游承受范围内把并发拉满。异步不是无限并发，要有背压。
六、效果数据
压测工具：wrk，4线程200连接，持续60秒，请求体固定10个user_id。
| 指标 | Before (Flask+Gunicorn) | After (FastAPI+uvicorn) | 提升 |
|---|---|---|---|
| QPS | 120 | 2300 | 19.2x |
| P50延迟 | 480ms | 42ms | 11.4x |
| P99延迟 | 1800ms | 95ms | 18.9x |
| CPU利用率 | 13% | 78% | - |
| 单机内存 | 320MB | 210MB | - |
机器配置：4核8G，CentOS 7.9，内网下游RT稳定在80ms。
成本角度：原来高峰期要12台机器扛，现在2台就够，一年省下来的云费用六位数。
七、总结
asyncio不是什么银弹，它的价值边界很清晰：IO密集、高并发、下游可并发。如果你的接口是CPU密集（比如图像处理、复杂计算），asyncio救不了你，得靠多进程或者换语言。
这次改造的三个核心经验：
- 连接池全局复用：AsyncClient、数据库连接池，一律单例
- 同步阻塞代码要隔离：
to_thread或run_in_executor，别阻塞事件循环 - 并发要有上限：
limit_concurrency、连接池大小、Semaphore，保护下游也是保护自己
最后提醒一句：asyncio的调试比同步代码痛苦，异常栈经常断在 gather 里。建议上 asyncio.TaskGroup（3.11+），异常传播清晰很多。
