1. 问题背景：一个“看起来没问题”的API
上周接手一个FastAPI项目，用户列表接口GET /api/v1/users?page=1&size=20，生产环境平均响应时间1200ms，P99高达3.2秒。最诡异的是，这个接口逻辑简单——查数据库、返回JSON，连复杂的计算都没有。
第一反应是“数据库慢”，但查了监控，MySQL CPU只有15%，连接数正常。用EXPLAIN看SQL，主查询走了主键索引，没毛病。
于是我开始怀疑代码层。翻看视图函数，发现一个典型问题：循环查询数据库。每个用户要查一次部门表、一次角色表、一次最近登录记录——这就是教科书级的N+1查询。
2. 环境与版本
- Python 3.11.4
- FastAPI 0.104.1
- Uvicorn 0.24.0（workers=4，--loop uvloop）
- SQLAlchemy 2.0.23（async模式）
- MySQL 8.0.33（InnoDB，buffer_pool_size=2G）
- Redis 7.0.12
- 压测工具：Locust 2.20.1
- 服务器：8C16G云主机，SSD
3. 第一刀：Profiling定位瓶颈
凭经验猜没用，先上工具。我写了个简单的profile脚本，用cProfile抓取一次请求的调用栈，并输出耗时前20的函数：
# profile_api.py
import cProfile
import pstats
import asyncio
from app.main import app
async def call_api():
from httpx import AsyncClient
async with AsyncClient(app=app, base_url="http://test") as client:
resp = await client.get("/api/v1/users?page=1&size=20")
assert resp.status_code == 200
def main():
pr = cProfile.Profile()
pr.enable()
asyncio.run(call_api())
pr.disable()
stats = pstats.Stats(pr)
stats.sort_stats("cumulative").print_stats(20)
if __name__ == "__main__":
main()
运行结果让我眉头一皱：
ncalls tottime percall cumtime percall filename:lineno
21 0.0023 0.0001 0.8921 0.0425 sqlalchemy/orm/loading.py:113
21 0.0011 0.0001 0.7432 0.0354 app/repositories/user_repo.py:45
60 0.0008 0.0000 0.6120 0.0102 sqlalchemy/orm/strategies.py:678
一目了然——60次子查询，每次累积耗时约0.01秒，总计0.6秒。这就是N+1的实锤。user_repo.py:45是那个循环查部门表的代码。
4. 第二刀：SQLAlchemy N+1修复与复合索引
问题代码长这样：
# 修复前：N+1查询
async def get_users_with_details(db: AsyncSession, page: int, size: int):
result = await db.execute(select(User).offset((page-1)*size).limit(size))
users = result.scalars().all()
user_list = []
for user in users:
# 每个用户都发起一次额外查询
dept = await db.get(Department, user.dept_id)
roles = await db.execute(select(Role).join(UserRole, UserRole.role_id == Role.id)
.where(UserRole.user_id == user.id))
user_list.append({
"id": user.id,
"name": user.name,
"dept_name": dept.name if dept else None,
"roles": roles.scalars().all()
})
return user_list
修复方案：用selectinload一次性加载关联对象，避免循环查询：
# 修复后：一次性加载关联
from sqlalchemy.orm import selectinload
async def get_users_with_details(db: AsyncSession, page: int, size: int):
stmt = (select(User)
.options(selectinload(User.department),
selectinload(User.roles))
.offset((page-1)*size)
.limit(size)
.order_by(User.created_at.desc()))
result = await db.execute(stmt)
users = result.scalars().all()
return [user_to_dict(u) for u in users]
同时，给users表加了一个复合索引。原来只有主键索引，但查询总是ORDER BY created_at DESC LIMIT 20，MySQL需要filesort。加索引：
ALTER TABLE users ADD INDEX idx_created_at (created_at DESC);
-- 如果经常按dept_id过滤，再加一个联合索引
ALTER TABLE users ADD INDEX idx_dept_created (dept_id, created_at DESC);
改完跑一次profile，N+1消失，接口耗时从1200ms降到340ms。但还不够，目标是把P99压到200ms以内。
5. 第三刀：Redis缓存策略（两级缓存）
数据库查询已经从60次降到1次，但每次请求仍然要往返数据库。对于用户列表这种读多写少的接口，缓存是必须的。
我设计了两级缓存：
- 一级缓存：Redis，缓存整个列表的JSON序列化结果，TTL 60秒。
- 二级缓存：应用内TTLCache（30秒），减少Redis网络IO。
实现代码如下：
# cache.py
import json
import time
from functools import wraps
from cachetools import TTLCache
from app.core.redis import redis_client
# 应用内缓存：最多1000个key，TTL 30秒
local_cache = TTLCache(maxsize=1000, ttl=30)
CACHE_PREFIX = "api:v1:users"
def cache_json(key: str, ttl: int = 60):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
cache_key = f"{CACHE_PREFIX}:{key}"
# 1. 尝试本地缓存
if cache_key in local_cache:
return local_cache[cache_key]
# 2. 尝试Redis缓存
cached = await redis_client.get(cache_key)
if cached:
local_cache[cache_key] = cached
return cached
# 3. 真实执行查询
result = await func(*args, **kwargs)
json_str = json.dumps(result, default=str)
# 写入两级缓存
await redis_client.setex(cache_key, ttl, json_str)
local_cache[cache_key] = json_str
return json_str
return wrapper
return decorator
# 使用示例
@cache_json(key="page_{page}_size_{size}", ttl=60)
async def get_user_page(page: int, size: int):
# 实际查询逻辑，返回dict/list
...
踩坑提醒：缓存粒度别太大。我一开始缓存整个列表（包括dept_name等关联字段），但用户角色变更后数据不一致。后来加了主动失效机制：用户更新、删除、角色变更时，删除对应page的缓存。简单粗暴但有效。
6. 压测对比：从80 QPS到1100 QPS
用Locust模拟50并发用户，持续压测5分钟，对比三个阶段的指标：
| 阶段 | 平均响应时间 | P95 | P99 | QPS |
|---|---|---|---|---|
| 初始版本 | 1200ms | 2100ms | 3200ms | 82 |
| 修复N+1+索引 | 340ms | 520ms | 760ms | 310 |
| 加Redis+本地缓存 | 95ms | 140ms | 180ms | 1100 |
压测命令（Locust headless模式）：
locust -f load_test.py --headless -u 50 -r 10 -t 5m --host=http://localhost:8000
load_test.py核心部分：
from locust import HttpUser, task, between
class ApiUser(HttpUser):
wait_time = between(0.1, 0.5)
@task(5)
def get_users(self):
self.client.get("/api/v1/users?page=1&size=20")
@task(1)
def get_user_detail(self):
self.client.get("/api/v1/users/12345")
额外优化：Uvicorn的--loop uvloop比默认asyncio快约15%。同时打开HTTP keep-alive，压测时减少重复握手。
7. 总结与经验
这次调优的核心收获：
- 别猜，先profile。cProfile + pstats 30秒定位问题，比瞎改快10倍。
- ORM的N+1是隐形杀手。
selectinload比joinedload更适合一对多场景，避免结果集膨胀。 - 索引不是越多越好。这里
idx_dept_created只对特定过滤场景有用，生产环境要考虑写入成本。 - 缓存要有失效策略。Redis TTL + 主动删除 + 本地短期缓存，三层保障一致性。
- 压测要分阶段对比，每个优化单独验证效果，避免混在一起不知道怎么回滚。
如果你的API也有类似性能问题，建议按这个顺序排查：profile → 数据库查询 → 索引 → 缓存。不要一上来就加缓存，那会把问题掩盖掉。
后记：调优后这个接口在线上稳定运行两周，平均响应时间稳定在85-110ms，没有出现缓存雪崩或穿透问题。Redis内存占用增加约1.5MB（缓存60秒的20条用户数据），成本可忽略。
