1. 问题背景：一个“看起来没毛病”的聚合接口
上个月接手一个订单服务，核心接口是 GET /api/v1/orders/summary，前端要展示用户近30天的订单概览：订单列表、商品快照、优惠明细、物流状态。业务逻辑不复杂，就是三张表join。
但压测结果很打脸：50并发下，P99延迟800ms，数据库CPU 90%，应用服务器CPU只有30%。明显是IO等待，不是计算瓶颈。
先看环境：
Python 3.11.4
FastAPI 0.104.1 (uvicorn 0.24.0, workers=4)
SQLAlchemy 2.0.21 + asyncpg 0.28.0
PostgreSQL 14.5 (16核32G, SSD)
Redis 7.0 (单机)
2. 第一步：用Profiling确定瓶颈在哪
很多同学一上来就改代码，我建议先花10分钟做profile。性能调优的第一原则：不要猜，要测。
写了两个脚本，一个是cProfile静态分析，一个是py-spy动态抓取生产栈。
# profiling/cprofile_runner.py
import cProfile
import pstats
import asyncio
from app.main import app
from httpx import ASGITransport, AsyncClient
async def profile_endpoint():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
for _ in range(200): # 预热
await client.get("/api/v1/orders/summary?user_id=1001")
# 正式采集
for _ in range(500):
resp = await client.get("/api/v1/orders/summary?user_id=1001")
assert resp.status_code == 200
if __name__ == "__main__":
profiler = cProfile.Profile()
profiler.enable()
asyncio.run(profile_endpoint())
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative").print_stats(30)
关键输出摘录：
ncalls tottime percall cumtime percall filename:lineno
500 0.012 0.000 0.482 0.001 sqlalchemy/orm/loading.py:80
500 0.008 0.000 0.356 0.001 sqlalchemy/orm/strategies.py:310 # selectinload
1500 0.021 0.000 0.298 0.000 asyncpg/protocol/protocol.pyx:120 # DB roundtrip
500 0.015 0.000 0.187 0.000 app/services/order_service.py:45 # 序列化
结论非常清晰：数据库查询占了65%以上时间，且selectinload触发了额外的SQL（N+1的变种）。JSON序列化只占0.18s，不是主要矛盾。
再用py-spy抓生产环境的线程栈（注意：生产环境千万别用cProfile，开销太大）：
# 在容器内执行
py-spy dump --pid $(pgrep -f uvicorn | head -1) --duration 5
看到大量线程阻塞在asyncpg的wait_for_message上，进一步确认是IO瓶颈。
3. 第二步：数据库查询优化 —— 从5次SQL降到1次
原始代码用了SQLAlchemy的lazy='select'，导致查询订单后，访问每个订单的items和promotions都会触发一次异步SQL。30个订单 = 1 + 30 + 30 = 61次查询。
方案：改用selectinload一次性批量加载，同时调整索引。
# app/repositories/order_repo.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from sqlalchemy import select, func
from app.models import Order, OrderItem, Promotion, Logistics
from datetime import datetime, timedelta
async def get_order_summary(session: AsyncSession, user_id: int, days: int = 30):
"""核心优化点：一次查询加载所有关联数据"""
since = datetime.utcnow() - timedelta(days=days)
# 1. 主查询：只查订单，用selectinload批量加载关联对象
stmt = (
select(Order)
.options(
selectinload(Order.items), # 替代 lazy='select'
selectinload(Order.promotions),
selectinload(Order.logistics),
)
.where(
Order.user_id == user_id,
Order.created_at >= since,
Order.status.in_(['paid', 'shipped', 'completed'])
)
.order_by(Order.created_at.desc())
.limit(50)
)
result = await session.execute(stmt)
return result.scalars().unique().all()
同时给PostgreSQL加了覆盖索引（之前只有单列索引user_id）：
-- 数据库迁移脚本 V20231115__add_composite_index.sql
CREATE INDEX CONCURRENTLY IF NOT EXISTS
idx_orders_user_created_status
ON orders (user_id, created_at DESC, status)
INCLUDE (total_amount, currency);
-- 商品表和物流表也加上外键索引
CREATE INDEX IF NOT EXISTS idx_order_items_order_id ON order_items (order_id) INCLUDE (sku_id, quantity, price);
CREATE INDEX IF NOT EXISTS idx_logistics_order_id ON logistics (order_id) INCLUDE (tracking_no, status);
效果：SQL查询次数从61次降到1次（主查询）+ 3次（批量加载）。单次请求数据库耗时从380ms降到95ms。
4. 第三步：Redis三级缓存 —— 读多写少的场景就该这么搞
数据库优化后P99降到220ms，但还不够。分析压测数据发现，80%的请求都在读相同用户的数据（用户频繁刷新页面）。
缓存设计：
- L1：进程内缓存（functools.lru_cache）—— 100ms内过期，适合突发流量
- L2：Redis缓存 —— 5分钟过期，存JSON序列化后的完整响应
- L3：数据库 —— 兜底
核心实现：
# app/services/cache_service.py
import json
import redis.asyncio as redis
from functools import lru_cache
from app.core.config import settings
redis_client = redis.from_url(
settings.REDIS_URL,
encoding="utf-8",
decode_responses=True,
socket_connect_timeout=0.5, # 连接超时设短，避免拖垮主流程
socket_timeout=0.5,
max_connections=32,
)
# L2: Redis缓存 —— 带版本号和业务前缀，方便主动失效
async def get_cached_order_summary(user_id: int, days: int = 30):
cache_key = f"order_summary:v2:{user_id}:{days}"
try:
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
except (redis.RedisError, json.JSONDecodeError) as e:
# 缓存故障降级，记录日志但不阻塞主流程
logger.warning(f"Redis cache read failed: {e}")
return None
async def set_cached_order_summary(user_id: int, days: int, data: dict, ttl: int = 300):
cache_key = f"order_summary:v2:{user_id}:{days}"
try:
await redis_client.setex(cache_key, ttl, json.dumps(data, default=str))
except redis.RedisError as e:
logger.warning(f"Redis cache write failed: {e}")
# L1: 进程内缓存 —— 配合Redis的TTL，设置更短的过期时间
def l1_cache_key(user_id: int, days: int):
return (user_id, days)
@lru_cache(maxsize=512, typed=True)
def get_l1_cached_data(user_id: int, days: int):
# 注意：这里只存计算结果，真正的数据从L2或DB获取
# 实际项目中建议用ttl_cache，此处简化
return None
def invalidate_order_cache(user_id: int, days: int = 30):
"""订单变更时主动失效缓存"""
get_l1_cached_data.cache_clear() # 简化处理，实际应精确删除
cache_key = f"order_summary:v2:{user_id}:{days}"
asyncio.create_task(redis_client.delete(cache_key))
然后在FastAPI路由中加上缓存逻辑：
# app/api/routes/orders.py
from fastapi import APIRouter, Depends, HTTPException
from app.services.cache_service import get_cached_order_summary, set_cached_order_summary
from app.services.order_service import get_order_summary_data
router = APIRouter(prefix="/api/v1")
@router.get("/orders/summary")
async def get_orders_summary(
user_id: int = Query(..., ge=1),
days: int = Query(30, ge=1, le=90),
session: AsyncSession = Depends(get_db)
):
# 1. 先查缓存
cached = await get_cached_order_summary(user_id, days)
if cached:
# 命中缓存，直接返回（连DB都不查）
return {"data": cached, "source": "redis"}
# 2. 缓存未命中，查库并构造响应
db_data = await get_order_summary_data(session, user_id, days)
if not db_data:
raise HTTPException(status_code=404, detail="No orders found")
# 3. 写回Redis，TTL设为5分钟
await set_cached_order_summary(user_id, days, db_data.dict())
return {"data": db_data, "source": "database"}
5. 踩坑与优化：缓存穿透、序列化、连接池
坑1：缓存穿透。压测时发现用户ID随机生成，Redis永远不命中，全部打到DB。解决方案是布隆过滤器或者对空结果也缓存（设置短TTL 30秒）。
async def get_cached_or_empty(user_id: int, days: int):
data = await get_cached_order_summary(user_id, days)
if data is None:
# 缓存空结果，避免穿透
await set_cached_order_summary(user_id, days, {"empty": True}, ttl=30)
return data
坑2：JSON序列化慢。Pydantic v2的dict()比手动json.dumps慢30%。改成直接操作ORM对象，用json.dumps(..., default=str)手动序列化，省去Pydantic开销。
坑3：连接池耗尽。Redis连接池默认10，高并发下不够用。调大max_connections=32，而且别忘了decode_responses=True，否则每次都要bytes解码。
压测工具用wrk和locust双验证：
# wrk 压测命令
wrk -t8 -c200 -d60s --latency http://localhost:8000/api/v1/orders/summary?user_id=1001
6. 效果数据：调优前后对比
| 指标 | 调优前 | 调优后 | 提升幅度 |
|---|---|---|---|
| P99 延迟 | 820ms | 62ms | 92.4% ↓ |
| P50 延迟 | 450ms | 38ms | 91.5% ↓ |
| QPS (200并发) | 220 | 1280 | 481% ↑ |
| 数据库CPU | 90% | 18% | 80% ↓ |
| 应用CPU | 30% | 55% | 合理上升 |
| SQL查询次数/请求 | 61 | 4 | 93.4% ↓ |
| Redis命中率 | 0% | 87.3% | - |
最终架构：FastAPI (uvicorn 4 workers) + SQLAlchemy 2.0 async + PostgreSQL + Redis (L2缓存) + 进程内缓存 (L1)。
7. 总结与建议
这次调优的核心收获不是某个具体技术，而是调优方法论：
- 先profile再动手：cProfile/profiling工具帮你定位真实瓶颈，大多数人花80%时间优化了20%无关紧要的代码。
- 数据库永远第一优先：N+1查询是Python异步ORM的通病，用
selectinload或joinedload批量加载。 - 缓存是最后手段：先优化SQL，再加缓存。否则缓存里存的是垃圾数据，只会让系统更复杂。
- 注意缓存失效：订单状态变化时要主动删除Redis key + 清L1缓存，否则用户看到过期数据会投诉。
另外说句大实话：别一上来就上微服务、消息队列。这个接口单机就能扛1200 QPS，根本不需要分布式方案。先把基础优化做好，比啥都强。
如果有同学遇到类似问题，可以按这个思路一步步排查。代码都放在GitHub仓库（链接见评论区），欢迎交流。
