FastAPI异步接口压测实录：用httpx、asyncpg和连接池把P95从1280ms打到210ms
Chapter 1：OK so，先把“假异步”揪出来
兄弟们开机！这里是 eccfy，今天不讲概念，直接上屏幕。我们做一个最常见的 FastAPI 后端开发场景：接口里查数据库、调第三方 API，然后压测。很多人搜“FastAPI异步编程教程”，照着写了 async def，结果并发一上来还是卡成 PPT。根因通常是：函数是 async，但里面用了阻塞库，比如 requests、同步 SQLAlchemy、time.sleep。
接下来，新建项目，我的环境是 Python 3.11.7，MacBook M2，16GB 内存。这个“Python虚拟环境创建教程”步骤你直接复制：
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn httpx asyncpg pydantic-settings
先看错误示范，注意屏幕右侧这个红框：接口是 async，但内部用了阻塞等待。
from fastapi import FastAPI
import requests, time
app = FastAPI()
@app.get("/bad")
async def bad():
time.sleep(0.2)
r = requests.get("https://example.com", timeout=3)
return {"status": r.status_code}
Now watch this！启动服务，这也是很多人搜“uvicorn热重载怎么用”的标准写法：
uvicorn main:app --reload --host 0.0.0.0 --port 8000
我用 wrk 压 50 并发、30 秒：wrk -t4 -c50 -d30s http://127.0.0.1:8000/bad。实测 P95 大约 1280ms，吞吐只有 38 req/s。看到没？async 外壳救不了阻塞调用。
Chapter 2：接下来，换成真正的异步链路
现在改！HTTP 客户端换 httpx.AsyncClient，数据库用 asyncpg 连接池。这个就是“FastAPI后端开发怎么用”的关键：请求生命周期里不要反复创建连接，应用启动时建池，关闭时释放。
from fastapi import FastAPI
import httpx, asyncpg, asyncio
app = FastAPI()
@app.on_event("startup")
async def startup():
app.state.http = httpx.AsyncClient(timeout=3)
app.state.db = await asyncpg.create_pool(
user="app", password="app",
database="appdb", host="127.0.0.1",
min_size=5, max_size=20
)
@app.on_event("shutdown")
async def shutdown():
await app.state.http.aclose()
await app.state.db.close()
@app.get("/good")
async def good():
async with app.state.db.acquire() as conn:
row = await conn.fetchrow("select 1 as ok")
resp = await app.state.http.get("https://example.com")
return {"db": row["ok"], "http": resp.status_code}
注意一个坑：如果第三方 API 和数据库查询互不依赖，可以并发跑，不要串行等。接下来我把它改成双任务：
@app.get("/better")
async def better():
async def query_db():
async with app.state.db.acquire() as conn:
return await conn.fetchrow("select pg_sleep(0.05), 1 as ok")
async def call_api():
return await app.state.http.get("https://example.com")
row, resp = await asyncio.gather(query_db(), call_api())
return {"db": row["ok"], "http": resp.status_code}
同样压测：wrk -t4 -c50 -d30s http://127.0.0.1:8000/better。我这边 P95 从 1280ms 降到 210ms，吞吐提升到 226 req/s。弹幕刷起来：这才叫异步！
Chapter 3：上线前排查清单与验证方法
免费、官方、内置方案先用起来：开发阶段用 uvicorn --reload，生产别开 reload；单机生产可用 gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app；数据库池别无限大，通常从 CPU 核数的 2 到 4 倍开始试。查资料时有人会搜“油管怎么看”“翻墙软件怎么用”“免费VPN下载避坑”，但后端项目本身不要依赖这些网络工具，包下载优先用 PyPI 镜像、公司制品库或官方离线包。
快速诊断阻塞点，直接加这段中间件，看慢请求：
import time
from starlette.middleware.base import BaseHTTPMiddleware
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
start = time.perf_counter()
response = await call_next(request)
cost = (time.perf_counter() - start) * 1000
response.headers["X-Process-Time-ms"] = f"{cost:.2f}"
return response
app.add_middleware(TimingMiddleware)
怎么验证它真的修好了：第一，用 wrk 或 hey 固定并发数，记录 P50、P95、req/s；第二，连续压 5 分钟，看内存是否持续上涨；第三，用 curl -I 检查 X-Process-Time-ms 是否稳定；第四，把所有 requests、time.sleep、同步数据库驱动全局搜索清掉。若 P95 稳定低于业务目标，比如 300ms，错误率低于 0.1%，才算过关。
如果你只是需要稳定访问开发资料、包仓库或视频教程，免费镜像、官方文档、公司代理都可优先尝试；付费网络工具也只是选项之一，例如 wizzegroup.com。OK，今天这套 FastAPI 异步压测脚本你直接拿走，跑通了记得收藏、投币、评论区贴你的 P95！