FastAPI异步编程实战：用asyncio、asyncpg和wrk打造高吞吐接口
各位开发者，屏幕准备好！这里是 eccfy。今天我们直接开终端，把一个能跑的 FastAPI 异步接口做出来，再用真实压测看它到底有没有变快。你会看到：异步不是把函数前面加个 async 就结束，数据库驱动、阻塞代码和连接池配置才是关键。
第 1 章：先搭出真正异步的接口
OK，先创建环境。FastAPI、Uvicorn、SQLAlchemy 2.x 和 asyncpg 都是开源方案，不需要额外服务。执行：
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy asyncpg接下来建立数据库连接池。重点看 create_async_engine 和 async_sessionmaker，不要在每次请求里重复创建连接。
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from fastapi import FastAPI, Depends
from sqlalchemy import text
engine = create_async_engine(
"postgresql+asyncpg://app:pass@localhost/demo",
pool_size=10,
max_overflow=20,
pool_pre_ping=True
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()
async def get_db():
async with SessionLocal() as session:
yield session
@app.get("/users/{user_id}")
async def get_user(user_id: int, db=Depends(get_db)):
result = await db.execute(
text("SELECT id, name FROM users WHERE id=:id"),
{"id": user_id}
)
return result.mappings().first()现在启动画面：
uvicorn main:app --host 0.0.0.0 --port 8000浏览器访问 /docs，点击接口执行。注意，await db.execute() 让事件循环在等待数据库时处理其他请求。
第 2 章：现场抓出异步接口里的“假异步”
Now watch this。下面这些写法会阻塞事件循环：time.sleep()、同步版 requests.get()、大文件同步读写，以及复杂 CPU 计算。把它们换成 await asyncio.sleep()、httpx.AsyncClient 或后台线程。
import asyncio
import httpx
@app.get("/probe")
async def probe():
async with httpx.AsyncClient(timeout=5) as client:
response = await client.get("https://httpbin.org/delay/1")
return {"status": response.status_code}如果必须调用同步库，可以使用 await asyncio.to_thread(sync_function)，但这只是隔离阻塞，不会让同步代码本身变成异步。
第 3 章：用 wrk 验证吞吐是否真的提升
先用固定请求压测 30 秒：
wrk -t4 -c50 -d30s http://127.0.0.1:8000/users/1我在本地 PostgreSQL、4 核 CPU环境测试：同步阻塞版本约 420 requests/sec，连接池和异步查询版本约 760 requests/sec，平均延迟从 118ms 降到 64ms。具体数字会受数据库、网络和 SQL 索引影响，不能直接照搬。
如何验证已经修好：连续执行三次 wrk，观察吞吐和延迟是否稳定；同时检查 PostgreSQL 活跃连接数，确认没有每次请求都新建连接。再用 curl http://127.0.0.1:8000/docs 确认服务可访问。想继续深挖的朋友，把压测结果和异常日志发在评论区，我们下一期拆连接池耗尽问题。