一、问题背景
去年底接手了一个聚合查询服务，逻辑很简单：前端传一个商品ID，后端要同时去三个上游服务拿数据——价格服务、库存服务、评价服务，然后拼成一个JSON返回。
原来的实现是Flask + requests，串行调用：
# before: app.py (Flask + requests)
import requests
from flask import Flask, jsonify, request
app = Flask(__name__)
PRICE_URL = "http://price-svc.internal/api/price"
STOCK_URL = "http://stock-svc.internal/api/stock"
REVIEW_URL = "http://review-svc.internal/api/review"
@app.route("/api/product/")
def get_product(pid):
price = requests.get(PRICE_URL, params={"pid": pid}, timeout=2).json()
stock = requests.get(STOCK_URL, params={"pid": pid}, timeout=2).json()
review = requests.get(REVIEW_URL, params={"pid": pid}, timeout=2).json()
return jsonify({"pid": pid, "price": price, "stock": stock, "review": review})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
三个上游平均响应分别是40ms、55ms、70ms，串行加起来165ms。听起来还行？但Flask默认的多线程模型在200并发下直接崩了——每个请求占一个worker线程，线程池打满后请求排队，P99延迟飙到2.4秒，QPS只有83。运维那边说机器已经加到8台了还是扛不住大促。
问题很清楚：这是典型的IO密集型场景，同步阻塞模型下线程全在等网络，CPU几乎是闲的。换成asyncio才是正解。
二、环境与版本
- Python 3.11.6（3.10+对asyncio的TaskGroup、异常组支持更好，建议别用3.8）
- aiohttp 3.9.3
- uvloop 0.19.0（Linux下必装，性能提升明显）
- FastAPI 0.109.2 + uvicorn 0.27.0（替代Flask）
- 压测工具：wrk 4.2.0，命令
wrk -t8 -c200 -d60s http://.../api/product/123
三、方案设计
核心思路就两条：
- 用异步HTTP客户端替换requests：aiohttp的
ClientSession支持连接池复用，避免每次请求重建TCP。 - 用
asyncio.gather并发调用三个上游：总耗时从"三个之和"变成"三个的最大值"。
另外几个细节：
- 全局复用一个ClientSession，不要每个请求都async with ClientSession()，否则连接池白搭。
- 给每个上游单独设超时，用asyncio.wait_for包一层，避免一个慢上游拖垮整个请求。
- 用uvloop替换默认事件循环，实测QPS能再涨20%左右。
- 用FastAPI替代Flask，因为它原生支持async def路由，Flask即使跑在asyncio上也是伪异步。
四、核心实现
after: main.py (FastAPI + aiohttp + uvloop)
# after: main.py
import asyncio
import aiohttp
import uvloop
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
PRICE_URL = "http://price-svc.internal/api/price"
STOCK_URL = "http://stock-svc.internal/api/stock"
REVIEW_URL = "http://review-svc.internal/api/review"
# 连接池配置：单host最大100连接，全局上限200
CONNECTOR = aiohttp.TCPConnector(
limit=200,
limit_per_host=100,
ttl_dns_cache=300,
keepalive_timeout=30,
)
TIMEOUT = aiohttp.ClientTimeout(total=1.5, connect=0.3)
session: aiohttp.ClientSession | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global session
session = aiohttp.ClientSession(connector=CONNECTOR, timeout=TIMEOUT)
yield
await session.close()
app = FastAPI(lifespan=lifespan)
async def fetch(url: str, pid: str) -> dict:
try:
async with session.get(url, params={"pid": pid}) as resp:
resp.raise_for_status()
return await resp.json()
except asyncio.TimeoutError:
raise HTTPException(504, f"upstream timeout: {url}")
except aiohttp.ClientError as e:
raise HTTPException(502, f"upstream error: {e}")
@app.get("/api/product/{pid}")
async def get_product(pid: str):
# 三个上游并发，谁先回来谁先算
price, stock, review = await asyncio.gather(
fetch(PRICE_URL, pid),
fetch(STOCK_URL, pid),
fetch(REVIEW_URL, pid),
return_exceptions=False,
)
return {"pid": pid, "price": price, "stock": stock, "review": review}
启动命令：
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop
这里--workers 4是因为机器是4核，一般建议 workers = CPU核数，再多反而因为进程切换开销掉性能。
关键点拆解
为什么全局一个session？
aiohttp的ClientSession内部维护连接池，如果每个请求都new一个，等于每次都重新建TCP+TLS，延迟和FD都会爆炸。我第一次压测忘了这点，QPS只有300，还伴随大量Too many open files。
为什么要limit_per_host=100？
默认值是0（无限制），在200并发下会瞬间打满上游的连接数，上游直接限流。设成100意味着单host最多100个复用连接，超出排队，反而更稳。
为什么用asyncio.gather而不是asyncio.wait？
gather直接返回结果列表，写法干净；wait返回done/pending集合，适合需要部分结果或动态取消的场景。我们这个场景是"三个都要，缺一不可"，gather足够。
五、踩坑与优化
坑1：return_exceptions=False时的异常传播
gather默认return_exceptions=False，任何一个子任务抛异常会立刻冒泡到调用方，其他未完成的任务不会被取消——它们还在后台跑。这会导致异常路径下连接泄漏。稳妥写法是配合TaskGroup（Python 3.11+）：
async def get_product_v2(pid: str):
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(fetch(PRICE_URL, pid))
t2 = tg.create_task(fetch(STOCK_URL, pid))
t3 = tg.create_task(fetch(REVIEW_URL, pid))
return {"pid": pid, "price": t1.result(), "stock": t2.result(), "review": t3.result()}
TaskGroup在任何子任务失败时会自动取消其余任务，语义更符合"要么全成功，要么全失败"。
坑2：CPU密集逻辑别塞进async函数
路由里有个签名校验，原来用的是hmac+JSON序列化，几十微秒的事，我一开始没在意。后来压测发现QPS上不去，用py-spy一采样，发现事件循环被这个同步调用卡住了。解决办法：要么扔到asyncio.to_thread，要么换更快的序列化库。小逻辑无所谓，但一旦超过100微秒就要警惕阻塞事件循环。
坑3：uvloop和某些C扩展不兼容
uvloop在Linux下性能提升约20-30%，但它替换了默认事件循环，某些依赖asyncio子类行为的库（比如某些老版本的aiomysql）会出问题。我们只用了aiohttp和httpx，没踩到，但升级前建议跑一遍集成测试。
坑4：DNS缓存
ttl_dns_cache=300这个参数很关键。默认是0，意味着每次请求都走一次DNS解析——内网DNS虽然快，但200并发下也扛不住。设成300秒后，DNS开销基本归零。
六、效果数据
压测环境：4核8G容器，上游服务在同一个K8s集群内网。wrk参数统一 -t8 -c200 -d60s。
| 版本 | 部署方式 | QPS | P50 | P99 | 错误率 |
|---|---|---|---|---|---|
| before (Flask+requests) | 8台×4worker | 83 | 1.2s | 2.4s | 0.3% |
| 中间版 (FastAPI+aiohttp，无uvloop，无连接池) | 1台×4worker | 310 | 380ms | 1.1s | 0.1% |
| after (FastAPI+aiohttp+uvloop+连接池) | 1台×4worker | 1100 | 165ms | 210ms | 0% |
几个关键观察：
- 从同步改异步，QPS直接×3.7（83→310），这一步收益最大。
- 加上uvloop和连接池参数调优，又×3.5（310→1100）。
- P99从2.4s降到210ms，是因为并发调用把串行165ms变成了max(40,55,70)=70ms，再加上异步调度和连接复用。
- 机器数从8台缩到2台（留一台做HA），省了6台。
CPU使用率从原来的15%涨到65%——这才是IO密集型服务该有的样子，之前CPU基本在睡觉。
七、总结
这个案例其实很典型：IO密集型服务用同步模型就是浪费机器。换成asyncio后，代码量没增加多少，性能却差了十几倍。
几条经验：
- 先看场景再选方案。如果上游是数据库且驱动不支持异步（比如老版本pymysql），硬上asyncio反而更慢，不如用
to_thread或者干脆保持同步+多进程。 - 连接池是异步HTTP的命脉。不做连接复用，异步的收益会打对折。
- uvloop在Linux上是白送的20%，Windows下没有，注意环境。
- 别在async函数里做CPU密集操作，事件循环一卡，所有并发都是假的。
- 压测数据要跑够60秒，短时间的QPS有连接预热的水分。
如果你的服务也是"调多个外部接口然后聚合返回"这种模式，强烈建议试试asyncio，投入产出比非常高。
