一、问题背景：一个“简单”订单接口为何拖垮整个服务？
上周三上线了一个订单列表API，功能很简单：根据用户ID查询最近30天订单，返回商品名称、价格和状态。测试环境一切正常，但压测环境一跑，监控面板立刻亮红灯——P95延迟2.3秒，错误率飙到8%，数据库连接池被打满。
我第一反应是“又是慢SQL”，但查了慢查询日志发现并没有特别离谱的语句。用top看CPU，发现Python进程占用高达380%，这不对劲。于是我开始系统性地定位瓶颈。
二、环境与版本：调优前的“家底”
先交代一下生产环境配置，避免大家踩同样的坑：
- Python 3.11.5（注意：3.11的字典和JSON序列化比3.10快约20%，但仍有优化空间）
- FastAPI 0.104.1（Uvicorn 0.24.0，单Worker）
- Flask 3.0.0（Gunicorn 21.2.0，3个Worker）
- PostgreSQL 15.3（单机，32GB内存，SSD）
- Redis 7.2（单实例，用于缓存）
- 压测工具：wrk 4.2.0，
-t4 -c100 -d30s（4线程，100并发，持续30秒） - ORM：SQLAlchemy 2.0.21（async模式）
调优前基线数据（同一台压测机，同一数据集）：
| 指标 | FastAPI | Flask |
|---|---|---|
| QPS | 152 | 98 |
| P95延迟 | 2.3s | 3.1s |
| CPU占用 | 380% | 290% |
三、Profiling定位：别猜，用工具说话
我用了两个工具：py-spy（无侵入式，适合生产环境）和cProfile（本地详细分析）。
第一步：py-spy dump 抓取CPU占用高的线程：
# 找到进程ID
$ py-spy top --pid 12345 --duration 10
输出片段：
%Own %Total OwnTime TotalTime Function
45.2% 45.2% 4.5s 4.5s json.dumps (json/encoder.py:199)
32.1% 77.3% 3.2s 3.2s sqlalchemy.orm.session.execute (session.py:171)
12.4% 89.7% 1.2s 1.2s asyncpg.connect (protocol.py:84)
看到没？JSON序列化占45%，SQLAlchemy执行占32%。这解释了我的疑问——不是单纯慢SQL，而是序列化和ORM开销过大。
第二步：cProfile 定位具体函数，代码里跑一下：
import cProfile
import pstats
from app.main import app
from fastapi.testclient import TestClient
client = TestClient(app)
with cProfile.Profile() as pr:
for _ in range(100):
client.get("/orders?user_id=123")
stats = pstats.Stats(pr)
stats.sort_stats("cumulative").print_stats(20)
关键输出：
ncalls tottime percall cumtime percall filename:lineno(function)
100 0.002 0.000 12.345 0.123 app/api/orders.py:15(get_orders)
100 0.001 0.000 8.234 0.082 app/services/order_service.py:22(fetch_orders)
1200 0.003 0.000 7.891 0.006 app/repositories/order_repo.py:45(_query_orders)
1200 0.001 0.000 6.234 0.005 sqlalchemy/orm/loading.py:198(load_on_ident)
核心发现：fetch_orders内部有1200次SQL查询——用户查了100次，每次返回30个订单，每个订单又查商品表（N+1问题）。加上SQLAlchemy默认的lazy="select"，每个订单商品都要单独发一条SQL。
四、方案设计：三步走
- 优化ORM查询：用
selectinload一次性加载关联表，把N+1变成两次SQL。 - 加Redis缓存：热点用户（活跃买家）的订单列表缓存30秒，减少数据库压力。
- 调整部署方式：FastAPI用Gunicorn+Uvicorn多Worker，Flask确保Gunicorn配置正确。
五、核心实现：代码与配置
5.1 SQLAlchemy查询优化
优化前（罪魁祸首）：
# order_repo.py
async def fetch_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()
# 这里每个order都会触发一次商品查询！
for order in orders:
await session.refresh(order, ["items"])
return orders
优化后：
# order_repo.py
from sqlalchemy.orm import selectinload
async def fetch_orders_optimized(user_id: int):
async with async_session() as session:
stmt = (
select(Order)
.where(Order.user_id == user_id)
.options(
selectinload(Order.items), # 一次性join加载
selectinload(Order.items).joinedload(Item.product) # 如果还有嵌套
)
.limit(30) # 明确限制数量
)
result = await session.execute(stmt)
return result.scalars().unique().all()
为什么快：selectinload会生成WHERE order_id IN (...)的查询，把30次查询合并为1次批量查询。数据库往返次数从31次降为2次。
5.2 Redis缓存策略
实现（FastAPI和Flask通用思路）：
# cache_service.py
import json
import redis.asyncio as aioredis
from fastapi import Request
redis_client = aioredis.from_url(
"redis://localhost:6379",
encoding="utf-8",
decode_responses=True,
max_connections=20
)
async def get_cached_orders(user_id: int):
cache_key = f"orders:{user_id}"
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
orders = await fetch_orders_optimized(user_id)
# 只缓存热点用户（比如最近7天有订单的用户）
await redis_client.setex(cache_key, 30, json.dumps(orders))
return orders
踩坑：千万别缓存所有用户！冷用户的数据会浪费内存，且缓存命中率低。我加了个逻辑：只有该用户近7天有订单才缓存，命中率从12%提升到67%。
5.3 部署配置对比
FastAPI（Gunicorn + Uvicorn Worker）：
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 --timeout 60 --max-requests 1000 \
--max-requests-jitter 100 --worker-connections 1000
Flask（Gunicorn同步Worker + 关闭debug）：
gunicorn app:app -w 4 --bind 0.0.0.0:5000 \
--worker-class sync --timeout 60 --max-requests 2000
关键参数解释：
- -w 4：4个Worker进程，充分利用多核CPU（压测机是4核）。
- --max-requests：每个Worker处理1000个请求后重启，防止内存泄漏（Python的长期运行内存增长问题）。
- --worker-connections：Uvicorn异步Worker的并发连接数。
六、踩坑与二次优化
坑1：Redis连接池打爆。加了缓存后，压测一开始Redis就报Connection pool exhausted。原因是我没设置max_connections，默认只有10。调大后解决。
坑2：JSON序列化还是慢。即使数据库查询优化了，json.dumps依然占CPU约25%。我改用orjson（用Rust编写的JSON库，比标准库快3-5倍）：
import orjson
def serialize_orders(orders):
# 直接返回bytes，FastAPI会自动处理
return orjson.dumps(
[{"id": o.id, "items": [i.name for i in o.items]} for o in orders]
)
坑3：Flask的同步Worker在长查询时阻塞。虽然Gunicorn有4个Worker，但每个Worker是单线程的，如果查询耗时1秒，该Worker就卡住。解决方法是把Flask的SQLAlchemy改为异步（或加gevent worker），但我用了个取巧办法：把耗时操作丢给线程池：
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=8)
@app.route("/orders")
def get_orders():
user_id = request.args.get("user_id")
future = executor.submit(fetch_orders_blocking, user_id)
orders = future.result(timeout=5)
return jsonify(orders)
七、效果数据：调优前后对比
压测条件：wrk -t4 -c100 -d30s，模拟100个并发用户，每个用户查询最近30天订单。
| 指标 | FastAPI调优前 | FastAPI调优后 | Flask调优前 | Flask调优后 |
|---|---|---|---|---|
| QPS | 152 | 748 | 98 | 412 |
| P95延迟 | 2.3s | 183ms | 3.1s | 520ms |
| 平均延迟 | 1.1s | 95ms | 1.8s | 210ms |
| 数据库查询次数/请求 | 31 | 2 | 31 | 2 |
| Redis缓存命中率 | 0% | 67% | 0% | 61% |
| CPU占用（峰值） | 380% | 220% | 290% | 180% |
额外观察：
- FastAPI调优后QPS提升392%，P95延迟下降92%。
- Flask调优后QPS提升320%，但P95延迟仍有520ms——为什么？因为Flask的同步模型在100并发下，4个Worker各处理25个请求，每个请求如果涉及IO等待（Redis/DB），Worker就被占用。后来我试了gevent worker，延迟降到220ms，但稳定性不如Uvicorn。
八、总结与建议
这次调优让我深刻体会到：性能瓶颈往往是“组合拳”而不是单一原因。我一开始只盯着SQL，结果发现JSON序列化占了大头。建议各位开发者：
- 先用py-spy抓现场，别猜。生产环境用
py-spy dump看热函数，开发环境用cProfile。 - ORM的N+1查询是默认陷阱，所有关联对象都要显式用
selectinload或joinedload。 - 缓存不是银弹，只缓存热点数据，否则内存和Redis连接都是瓶颈。
- 部署方式影响巨大：FastAPI必须用Uvicorn的异步Worker，Flask用Gevent或异步化改造，否则多Worker也扛不住IO密集。
最后，调优后别忘了压测回归——我调优完第二天，业务方说要加一个“导出Excel”功能，差点又崩了。下一篇文章我会分享大文件下载API的流式传输优化，敬请关注。
（完）
