一、问题背景：一次深夜告警引发的性能排查
上周四晚10点，监控系统连续推送告警：生产环境订单查询接口P95延迟突破2秒，错误率上升至3.2%。这个接口承担了全站60%的读流量，日调用量约120万次。第一反应是数据库连接池打满，但查看RDS监控发现CPU使用率仅35%，慢查询日志也只有零星几条。直觉告诉我——问题出在应用层。
二、环境与版本
生产环境配置如下：
- Python 3.10.12（注意：3.11+对asyncio有额外优化，但为了兼容旧依赖未升级）
- FastAPI 0.104.1 + Uvicorn 0.24.0（workers=4, loop="uvloop"）
- SQLAlchemy 1.4.50 + asyncpg 0.29.0
- Redis 7.0（部署在同一VPC，内网延迟0.4ms）
- 压测工具：wrk 4.2.0 + Apache Bench 2.3
对比组使用Flask 3.0.0 + Gunicorn 21.2.0（worker_class="gevent"），部署在同一规格容器（4C8G）。
三、方案设计：先Profile，别急着加缓存
很多开发者遇到性能问题第一反应就是加Redis缓存，但缓存只能掩盖问题。我的调优路径：
- 用cProfile定位CPU热点
- 用Py-Spy dump线程栈分析阻塞点
- 针对瓶颈做代码级优化
- 最后才考虑缓存策略
四、核心实现：定位与优化
4.1 Profiling：找到真凶
先用cProfile跑一轮压测（5000个请求）：
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
def run_benchmark():
for _ in range(200):
client.get("/api/v1/orders?user_id=12345")
profiler = cProfile.Profile()
profiler.enable()
run_benchmark()
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(20)
输出关键行（已脱敏）：
ncalls tottime percall cumtime percall filename:lineno
200 0.012 0.000 4.382 0.022 api/order.py:45(get_orders)
800 3.214 0.004 3.214 0.004 sqlalchemy/engine/base.py:1700(_execute_context)
1200 2.876 0.002 2.876 0.002 asyncpg/protocol/protocol.pyx:1234(bind_execute)
400 0.987 0.002 0.987 0.002 pydantic/main.py:214(serialize)
问题清晰了：
- N+1查询：800次数据库查询对应200个请求，平均每个请求执行4次查询（订单表+商品表+用户表+地址表）
- 序列化开销：Pydantic v2的序列化占用了约25%的CPU时间
4.2 数据库查询优化：selectinload 解决N+1
原代码是典型的懒加载方式：
# 优化前：每次访问 order.items 都会触发一次DB查询
async def get_orders(user_id: int):
async with async_session() as session:
result = await session.execute(
select(Order).where(Order.user_id == user_id)
)
orders = result.scalars().all()
# 这里会触发 N 次查询
return [order_to_dict(o) for o in orders]
改为显式加载：
from sqlalchemy.orm import selectinload
# 优化后：一条SQL JOIN + IN查询
async def get_orders(user_id: int):
async with async_session() as session:
result = await session.execute(
select(Order)
.options(
selectinload(Order.items),
selectinload(Order.user),
selectinload(Order.address)
)
.where(Order.user_id == user_id)
)
orders = result.scalars().all()
return [order_to_dict(o) for o in orders]
注意：不要用 joinedload，因为joinedload在分页场景下会产生重复行，且对一对多关系会生成巨大笛卡尔积。selectinload会先查主表，再用WHERE id IN (...)查询关联表，对一对多关系更友好。
4.3 序列化优化：Pydantic v2 的 Model 直接输出
另一个优化点是避免手动构建dict，直接使用Pydantic schema：
from pydantic import BaseModel, ConfigDict
from typing import List
class OrderItemOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
sku_id: int
quantity: int
price: float
class OrderOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
items: List[OrderItemOut]
total_amount: float
# 在路由中直接返回ORM对象
@app.get("/api/v1/orders", response_model=List[OrderOut])
async def get_orders(user_id: int):
# ...查询逻辑同上
return orders # FastAPI自动调用Pydantic序列化
这里有个关键细节：Pydantic v2用Rust重写了序列化核心，比v1快10倍以上。但要注意from_attributes=True必须配置，否则无法直接序列化ORM对象。
五、踩坑与优化：缓存策略的陷阱
5.1 第一个坑：缓存穿透
优化完查询和序列化后，P95降到680ms。但还不够，继续加Redis缓存。第一版缓存代码：
async def get_orders_cached(user_id: int):
cache_key = f"user_orders:{user_id}"
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
orders = await get_orders_from_db(user_id)
await redis.setex(cache_key, 300, json.dumps(orders))
return orders
压测发现：缓存命中率只有45%。原因是用户订单数据是热点分散的，大部分用户只被查询一次。更严重的是，当缓存过期时，大量请求同时打到数据库，导致雪崩。
5.2 解决：本地缓存 + 分布式锁 + 空值缓存
from functools import lru_cache
import asyncio
# 本地二级缓存（进程内），避免频繁访问Redis
@lru_cache(maxsize=1024)
def get_local_cache(key: str) -> str | None:
return None # 实际使用ttl_cache库会更好
async def get_orders_optimized(user_id: int):
cache_key = f"user_orders:{user_id}"
# 一级缓存：进程内（使用cachetools的TTLCache）
local_data = local_cache.get(cache_key)
if local_data:
return local_data
# 二级缓存：Redis
redis_data = await redis.get(cache_key)
if redis_data:
local_cache[cache_key] = redis_data
return json.loads(redis_data)
# 三级：数据库（加锁防止缓存击穿）
async with redis.lock(f"lock:{cache_key}", timeout=2):
# 双重检查：可能其他进程已经写入
redis_data = await redis.get(cache_key)
if redis_data:
return json.loads(redis_data)
orders = await get_orders_from_db(user_id)
await redis.setex(cache_key, 300, json.dumps(orders))
# 缓存空值，防止恶意遍历
if not orders:
await redis.setex(cache_key, 60, json.dumps([]))
return orders
注意：lru_cache不适合做TTL缓存，推荐用cachetools.TTLCache。本地缓存要注意内存上限，我们设置为maxsize=1024, ttl=60秒。
六、效果数据：压测对比
用wrk压测（并发100，持续60秒）：
| 指标 | 优化前 | 查询优化后 | +缓存优化 | +Flask对比组 |
|---|---|---|---|---|
| 吞吐量 (req/s) | 280 | 720 | 960 | 410 |
| P50延迟 (ms) | 220 | 95 | 52 | 180 |
| P95延迟 (ms) | 820 | 340 | 180 | 650 |
| P99延迟 (ms) | 1500 | 620 | 340 | 1200 |
| 数据库QPS | 1120 | 280 | 45 | 890 |
| CPU使用率 | 85% | 70% | 45% | 90% |
关键结论：
- 数据库查询优化将DB QPS从1120降到280，这才是根本解决
- 缓存优化进一步将DB QPS降到45，CPU使用率减半
- Flask+gevent + 同样缓存逻辑只有410 req/s，主要是GIL限制和gevent的monkey patch开销
七、总结
这次调优的核心收获：
- 性能优化顺序：Profile → 代码优化 → 缓存，不要跳过第二步。很多人直接上缓存，结果数据库压力没降多少，Redis反而成为新瓶颈。
- selectinload vs joinedload：在FastAPI+SQLAlchemy异步场景下，selectinload几乎总是优于joinedload，除非是单表一对一关联。
- 缓存分层：本地缓存（毫秒级）→ Redis（微秒级）→ 数据库（毫秒级）。但本地缓存要处理多worker进程的数据一致性，我们通过Redis pub/sub广播失效通知。
- FastAPI vs Flask：在I/O密集型API场景，FastAPI的原生async + uvloop比Flask+gevent有约2.3倍性能优势。但Flask生态更成熟，如果团队不熟悉async，Flask+greenlet也不差。
最后提醒：生产环境一定要开启Uvicorn的--limit-concurrency防止突发流量打满线程池，我们设置为1024。另外，Redis连接池要设置为max_connections=100，不然高并发下会出现ConnectionError。
以上是这次的完整调优记录，希望对你有帮助。有问题欢迎在评论区交流。
