FastAPI实战：从3秒到0.8秒的API性能优化手记

作者： /

刚开始我也以为FastAPI就是个带自动文档的Flask，直到真拿它做生产项目，连踩三个坑之后才明白：这玩意儿的设计哲学完全不同。今天不扯官方文档那些花架子，直接拿我最近改造的自动化运维平台说事儿。

先讲第一个坑：异步？骗人的吧？

我接手一个自动化任务调度API，原本用Flask写的，单个任务查询接口平均耗时3.2秒。同事说“换FastAPI，异步加持直接起飞”。我天真地信了，结果把def改成async def之后，耗时掉到3.1秒。

原因：FastAPI的异步只在你真正做IO操作时生效。如果你的async def里面还是同步的requests.get()或者time.sleep()，那它依然会阻塞事件循环。

看这个反例：
“python

❌ 假异步 - 依然阻塞

@app.get("/task/{task_id}")
async def get_task(task_id: str):
result = sync_db_query(f"SELECT * FROM tasks WHERE id={task_id}") # 同步阻塞
return {"data": result}
`

正确做法是用异步数据库驱动。我换成了databases库配合asyncpg：

`python

✅ 真异步 - 性能起飞

from databases import Database
database = Database("postgresql+asyncpg://user:pass@localhost/db")

@ap

p.get("/task/{task_id}")
async def get_task(task_id: str):
# 这里不会阻塞事件循环
query = "SELECT * FROM tasks WHERE id=:id"
result = await database.fetch_one(query=query, values={"id": task_id})
return {"data": result}
`

改完之后，单接口耗时直接从3.2秒降到0.8秒。注意，这不是算法优化，纯粹是IO模式改变。

但问题来了：如果你的数据库不支持异步怎么办？别急，后面有解决方案。

另一个坑：依赖注入的正确姿势

FastAPI的依赖注入系统很强大，但官方文档那段写的跟谜语一样。我花了三天才捋清楚最佳实践。

核心思想：把依赖注入当作函数组合，而不是类继承。

看这个登录鉴权的例子，我最开始写成这样：
`python

❌ 反面教材 - 把依赖写成装饰器

from fastapi import Depends, HTTPException

def verify_token(token: str = Header(...)):
if token != "secret":
raise HTTPException(status_code=401)
return token

@app.get("/protected")
def protected_route(token: str = Depends(verify_token)):
# 业务逻辑
pass
`

这个设计真的反人类：verify_token既做校验又返回token，耦合度太高。正确的做法是职责分离：

`python

✅ 正确的依赖注入模式

from fastapi import Depends, HTTPException, Header, Request
from typing import Optional

第一层：原始依赖

async def get_db() -> AsyncGenerator:
db = Database()
try:
await db.connect()
yield db
finally:
await db.disconnect()

第二层：业务依赖（基于原始依赖组合）

async def get_current_user(
request: Request,
db: Database = Depends(get_db)
) -> User:
token = request.headers.get("Authorization")
if not token:
raise HTTPException(status_code=401)

user = await db.fetch_one("SELECT * FROM users WHERE token=:token",
values={"token": token})
if not user:
raise HTTPException(status_code=403)
return User(**user)

第三层：业务接口 - 只用关心业务

@app.get("/profile")
async def get_profile(user: User = Depends(get_current_user)):
return user
`

这样拆的好处是：测试时可以轻松mock掉get_db，get_current_user也能独立测试。别问我怎么知道的——上线当天被QA追着打了三版才知道。

性能调优：那些数据说话的地方

注意：下面这些优化手段要在压测之后做，别没数据就瞎调。

我先贴一段压测脚本，用locust写的：
`python

性能测试脚本 - locustfile.py

from locust import HttpUser, task, between
import json

class APITestUser(HttpUser):
wait_time = between(1, 3)

@task(3) # 权重3
def get_tasks(self):
self.client.get("/api/tasks")

@task(1) # 权重1
def create_task(self):
payload = {"name": "test", "priority": 1}
self.client.post("/api/tasks", json=payload)
`

跑完第一次压测后，我发现一个奇怪的现象：QPS（每秒查询数）在200之后就上不去了，但CPU利用率只有40%。

根因：数据库连接池太小。

FastAPI默认的连接池大小是10，对于并发场景根本不够。看修复方案：
`python

调整数据库连接池

from databases import Database
import asyncpg

生产环境配置

database = Database(
"postgresql+asyncpg://user:pass@localhost/db",
min_size=5, # 最小连接数
max_size=50, # 最大连接数 - 原来默认10
max_queries=500, # 每个连接最大查询数
max_inactive_connection_lifetime=300 # 空闲连接超时
)
`

调完之后，同样负载下QPS从200飙到800，CPU利用率也上来了。

还有个技巧：响应模型规范

很多新手在定义API返回时，都喜欢这么写：
`python
@app.get("/tasks")
async def get_tasks():
tasks = await db.fetch_all("SELECT * FROM tasks")
return {"data": tasks, "total": len(tasks), "status": "ok"}
`

看着没问题对吧？错！这种返回结构没有类型约束，前端对接全靠猜。我踩过这个坑之后，强制自己用Pydantic的响应模型：

`python
from pydantic import BaseModel
from typing import List
from datetime import datetime

响应模型

class TaskOut(BaseModel):
id: int
name: str
status: str
created_at: datetime

class TaskListResponse(BaseModel):
data: List[TaskOut]
total: int
page: int
page_size: int

@app.get("/tasks", response_model=TaskListResponse)
async def get_tasks(page: int = 1, page_size: int = 10):
offset = (page - 1) * page_size
tasks = await db.fetch_all(
"SELECT * FROM tasks LIMIT :limit OFFSET :offset",
values={"limit": page_size, "offset": offset}
)
return TaskListResponse(
data=[TaskOut(**t) for t in tasks],
total=await db.fetch_val("SELECT COUNT(*) FROM tasks"),
page=page,
page_size=page_size
)
`

好处有三：

  • 自动生成API文档（Swagger里能看到完整的字段说明）
  • 接口返回格式强制统一（前端再也不用问“这个字段有没有”）
  • 参数校验自动生效（比如page必须是正整数）

    • 部署时的坑：别用默认的Uvicorn

    文档里说uvicorn main:app –reload开发用，但你敢在生产环境这么跑？我见过一个同事把–reload带到生产环境，CPU直接100%。

    生产环境正确部署方案：
    `python

    用Gunicorn + Uvicorn Workers

    gunicorn_config.py

    bind = "0.0.0.0:8000"
    workers = 4 # CPU核心数*2
    worker_class = "uvicorn.workers.UvicornWorker"
    keepalive = 65
    timeout = 120
    `

    然后启动：
    `bash
    gunicorn -c gunicorn_config.py main:app
    `

    这样单机就能抗住5000并发，比Uvicorn单进程强太多。

    免责声明：本文仅供参考，不构成医疗建议。

    总结一下，你可以立刻用的三个点

  • 异步改造：把同步DB驱动换成asyncpg+databases，同样的代码性能提升4倍以上
  • 依赖注入：用三层分离模式（原始依赖→业务依赖→业务接口），测试和维护都轻松
  • 部署规范：用gunicorn+uvicorn workers，别裸跑uvicorn`；数据库连接池调大到50

    • 最后说一句：FastAPI很好，但别把它当黑魔法。该压测还是得压测，该优化还得优化。工具只是武器，决定胜负的是你手里的活。

    滚动至顶部