一、问题背景：一个“还能用”的接口
事情是这样的，我们有个内部订单看板，前端每5秒轮询一次。上线初期数据量小，一切正常。但两个月后，订单表到了200万行，接口开始频繁超时（网关设置3s超时），监控面板上P95延迟飙到1.2秒。
我当时的内心：这接口不就是查个列表吗？怎么就这么慢？
先看原始代码（简化版），用的是FastAPI + SQLAlchemy 2.0 + PostgreSQL 14：
# app.py (调优前)
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session, joinedload
from database import get_db
from models import Order, Customer
app = FastAPI()
@app.get("/api/orders")
def get_orders(page: int = 1, size: int = 20, db: Session = Depends(get_db)):
orders = db.query(Order).order_by(Order.created_at.desc()) \
.offset((page-1) * size).limit(size).all()
result = []
for order in orders:
customer = db.query(Customer).filter(Customer.id == order.customer_id).first()
result.append({
"order_id": order.id,
"customer_name": customer.name if customer else "N/A",
"amount": order.amount,
"status": order.status_str() # 这个是个Python属性，内部有字符串拼接
})
return {"data": result, "total": db.query(Order).count()}
代码逻辑很简单：分页查订单，然后循环查客户。一眼就能看出问题——典型的N+1查询。但当时我天真地以为SQLAlchemy有缓存，不至于太慢。事实给了我一巴掌。
二、环境与版本信息
- Python 3.10.12
- FastAPI 0.104.1 + uvicorn 0.24.0（workers=4）
- SQLAlchemy 2.0.23 + psycopg2-binary 2.9.9
- PostgreSQL 14.5（16GB内存，SSD，单实例）
- Redis 7.0（用于缓存，后面会用到）
- 压测工具：Locust 2.19.1（单机8核8GB）
三、Profile定位：别猜，用工具说话
我不喜欢瞎猜性能瓶颈，直接上工具。
3.1 用py-spy抓在线调用栈
接口慢的时候，我直接对一个正在处理的worker进程做采样：
# 找到worker进程PID
ps aux | grep uvicorn
# 采样30秒，输出火焰图数据
py-spy record -p 12345 -o profile.svg --duration 30
火焰图出来后，我看了眼差点笑了——get_orders里那行db.query(Customer)占了接近70%的CPU时间。后面是status_str()这个Python方法（里面有个正则匹配），占了15%。
3.2 用cProfile做更细的追踪
py-spy是采样级别的，能看出热点分布，但我想看函数调用次数。于是用cProfile在本地跑一组真实请求：
# profile_test.py
import cProfile, pstats
from app import get_orders
from database import SessionLocal
def run_request():
db = SessionLocal()
try:
get_orders(1, 20, db)
finally:
db.close()
profiler = cProfile.Profile()
profiler.enable()
for _ in range(100): # 模拟100次请求
run_request()
profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative').print_stats(20)
结果关键输出：
ncalls tottime percall cumtime percall filename:lineno
100 0.002 0.000 3.812 0.038 app.py:12(get_orders)
2000 0.145 0.000 2.876 0.001 models.py:45(status_str)
2000 0.018 0.000 2.410 0.001 sqlalchemy/orm/query.py:2145(_execute_and_instances)
2000 1.980 0.001 2.200 0.001 sqlalchemy/dialects/postgresql/psycopg2.py:433(do_execute)
一目了然：2000次数据库查询（100次请求 × 每请求20个订单），每次查询平均1ms，但这2000次串行执行就变成了2秒。加上status_str的正则处理又吃掉0.8秒。总耗时3.8秒，跟线上P95差不多。
结论：瓶颈在N+1查询 + 低效的Python属性方法。
四、优化方案设计
针对定位到的问题，我做了三个方向的优化：
- 消除N+1：用
selectinload一次性加载关联客户，并把status_str改成预计算的状态字段（数据库里直接存字符串，而不是每次正则匹配）。 - 减少重复计算：订单列表的聚合统计（total）用Redis缓存，TTL 60秒。这个接口每5秒轮询一次，60秒内统计结果几乎不变。
- SQL层面优化：给
created_at加索引（其实之前有，但走了全表排序），改成ORDER BY created_at DESC LIMIT走索引扫描。
下面是优化后的核心代码：
# app_optimized.py (调优后)
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session, selectinload
from sqlalchemy import func, select
from redis import Redis
import json
from database import get_db
from models import Order, Customer
app = FastAPI()
redis_client = Redis(host='localhost', port=6379, db=0, decode_responses=True)
CACHE_TTL = 60 # 秒
@app.get("/api/orders")
def get_orders(page: int = 1, size: int = 20, db: Session = Depends(get_db)):
# 1. 缓存total统计，避免每次count全表
cache_key = "order:total"
total = redis_client.get(cache_key)
if total is None:
total = db.execute(select(func.count(Order.id))).scalar()
redis_client.setex(cache_key, CACHE_TTL, total)
else:
total = int(total)
# 2. 使用selectinload一次性加载客户，消除N+1
orders = db.execute(
select(Order)
.options(selectinload(Order.customer)) # 假设有relationship
.order_by(Order.created_at.desc())
.offset((page-1) * size)
.limit(size)
).scalars().all()
# 3. 直接读取预计算的status_str字段（不再调用Python方法）
result = [{
"order_id": o.id,
"customer_name": o.customer.name if o.customer else "N/A",
"amount": o.amount,
"status": o.status, # 数据库存储的字符串字段
} for o in orders]
return {"data": result, "total": total}
4.1 数据库迁移：加状态字段
原来status_str是Python属性，我改成在模型里加一个status字符串列，用一次数据迁移把旧数据更新：
# models.py 修改
class Order(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True)
# ...其他字段
status = Column(String(20), nullable=False, default="pending")
customer_id = Column(Integer, ForeignKey("customers.id"))
customer = relationship("Customer", lazy="raise") # 禁止懒加载，强制显式load
迁移SQL（用Alembic或直接执行）：
ALTER TABLE orders ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'pending';
CREATE INDEX ix_orders_created_at ON orders (created_at DESC);
-- 一次性回填：从旧逻辑计算status
UPDATE orders SET status = CASE WHEN ... END;
4.2 Flask版本对比
我顺便写了个Flask 3.0 + Flask-SQLAlchemy 3.1的版本做对比，核心区别在视图函数和session管理：
# flask_app.py
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, func
import redis
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:pass@localhost/db'
db = SQLAlchemy(app)
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
@app.get('/api/orders')
def get_orders():
page = request.args.get('page', 1, type=int)
size = request.args.get('size', 20, type=int)
total = redis_client.get('order:total')
if total is None:
total = db.session.execute(select(func.count(Order.id))).scalar()
redis_client.setex('order:total', 60, total)
else:
total = int(total)
# 同样使用selectinload
orders = db.session.execute(
select(Order).options(selectinload(Order.customer))
.order_by(Order.created_at.desc())
.offset((page-1) * size).limit(size)
).scalars().all()
result = [{
"order_id": o.id,
"customer_name": o.customer.name if o.customer else "N/A",
"amount": o.amount,
"status": o.status
} for o in orders]
return jsonify({"data": result, "total": total})
Flask这边性能几乎一致，主要差异在异步支持。FastAPI原生async，但SQLAlchemy同步查询会阻塞线程，所以实际性能在低并发下两者差不多。
五、踩坑与二次优化
坑1：selectinload没生效
我第一次改完，用py-spy一测，发现还是有N+1。原因是我在Order模型里定义了customer = relationship("Customer", lazy="select")，默认懒加载会覆盖selectinload的选项？其实不会，问题是options(selectinload(Order.customer))写错了，我写成了joinedload。后来换成selectinload就对了。
注意：joinedload在分页+order by场景下会有性能问题（LEFT OUTER JOIN + DISTINCT），所以这里用selectinload更合适。
坑2：Redis缓存穿透
第一次压测时，我把TTL设成了300秒，结果压测并发一上来，缓存刚好过期，所有请求同时打DB，数据库CPU瞬间100%。后来改成互斥锁重建策略：
def get_total_with_lock(db):
cache_key = "order:total"
total = redis_client.get(cache_key)
if total is not None:
return int(total)
# 加锁，只让一个请求重建缓存
lock_key = "order:total:lock"
if redis_client.set(lock_key, "1", nx=True, ex=5):
try:
total = db.execute(select(func.count(Order.id))).scalar()
redis_client.setex(cache_key, CACHE_TTL, total)
return total
finally:
redis_client.delete(lock_key)
else:
# 没拿到锁，等待100ms后重试
time.sleep(0.1)
return get_total_with_lock(db)
坑3：COUNT(*)依然慢
即使走了缓存，但如果缓存失效，SELECT COUNT(*) FROM orders在200万行上依然要300ms。我加了pg_stat_statements看执行计划，发现走了全表扫描。解决办法：用cron定期把count缓存在Redis里，或者接受60秒的陈旧数据。对于看板场景，完全没问题。
六、效果数据：压测对比
用Locust压测，模拟200并发用户，持续5分钟，数据如下：
| 指标 | 调优前 | 调优后 | 提升幅度 |
|---|---|---|---|
| 平均延迟 | 820ms | 88ms | 89.3% |
| P95延迟 | 1.2s | 105ms | 91.3% |
| QPS | 82 | 452 | 451% |
| 数据库CPU | 78% | 32% | -59% |
| 数据库查询数/请求 | 21 | 2 | -90.5% |
压测命令（Locust）：
locust -f locustfile.py --headless -u 200 -r 20 -t 5m --host http://localhost:8000
locustfile.py 简略版：
from locust import HttpUser, task, between
class OrderUser(HttpUser):
wait_time = between(1, 3)
@task
def get_orders(self):
self.client.get("/api/orders?page=1&size=20")
调优后，P95稳定在100ms左右，数据库连接池（SQLAlchemy默认pool_size=5，max_overflow=10）没有报警，Redis内存占用增加了约2MB（缓存一个count值而已）。
七、总结与TIPS
- 不要相信“ORM会自动优化”。SQLAlchemy的懒加载是性能杀手，必须显式用
selectinload或joinedload。 - Profile工具是刚需：py-spy能抓线上火焰图，cProfile能做本地详细分析。别猜，直接看数据。
- 缓存一定要考虑穿透和雪崩：互斥锁重建是简单有效的方案，比“永远不过期”更安全。
- FastAPI和Flask在这个场景下性能几乎无差异。选型别纠结，真正瓶颈在数据库查询和代码逻辑上。
- 最后留个问题：如果你的查询条件特别复杂（多字段过滤、排序），
selectinload可能不够，这时候考虑ES或ClickHouse。但大多数业务场景，把N+1干掉就够了。
以上就是完整调优过程，有问题评论区聊。
