一、问题背景：一个慢接口引发的血案
先交代下业务场景：一个商品列表API，需要返回商品基本信息、库存、最近3条评价。上线后用户反馈“划不动”，看了下监控，该接口P99延迟在800ms左右，数据库连接池经常被打满。我们当时用的是Flask 2.3 + SQLAlchemy 1.4 + MySQL 8.0，部署在4核8G的云服务器上。
这并不是一个复杂的接口，为什么这么慢？直觉告诉我，肯定不是业务逻辑复杂，而是代码“写得比较随意”。于是我开始用Profiling工具排查。
二、环境与版本
Python 3.10.12
Flask 2.3.3 → FastAPI 0.104.1
SQLAlchemy 1.4.50 → 2.0.23
PyMySQL 1.1.0
Redis 7.0.12 (单实例)
wrk 4.2.0 (压测工具)
MySQL 8.0.33 (buffer_pool_size=2G, innodb_io_capacity=2000)
压测命令统一使用：
wrk -t4 -c32 -d30s --latency http://localhost:5000/products
三、第一步：Profiling定位瓶颈
我先在Flask应用中插入了一个简单的profiling中间件，用cProfile采样，输出到KCacheGrind查看：
# profiling_middleware.py
import cProfile, pstats, io
def profile_request(app):
@app.before_request
def start_profile():
g.profiler = cProfile.Profile()
g.profiler.enable()
@app.after_request
def end_profile(response):
g.profiler.disable()
s = io.StringIO()
ps = pstats.Stats(g.profiler, stream=s).sort_stats('cumtime')
ps.print_stats(20)
app.logger.debug("Profile:\n%s", s.getvalue())
return response
结果是触目惊心的：
- sqlalchemy.orm.query.Query.all() 占了总耗时的63%
- json.dumps() 占12%
- copy.deepcopy() 占8%（这是历史遗留代码）
具体来说，商品列表返回了100条商品，每个商品又去数据库查询库存和评价，典型的N+1问题。而且评价查询用了order_by(create_time.desc()).limit(3)，但没加索引，导致文件排序。
四、数据库查询优化：从N+1到一次性加载+索引
方案设计：使用SQLAlchemy的selectinload一次性加载关联对象，同时给评价表的product_id和create_time加复合索引。
# 优化前：N+1查询
products = Product.query.filter(Product.status==1).all()
for p in products:
p.stock # 触发库存查询
p.reviews[:3] # 触发评价查询
# 优化后：一次性加载
from sqlalchemy.orm import selectinload
products = db.session.execute(
select(Product)
.options(selectinload(Product.stock))
.options(selectinload(Product.reviews))
.where(Product.status == 1)
.order_by(Product.create_time.desc())
).scalars().all()
同时给评价表加索引：
ALTER TABLE reviews ADD INDEX idx_product_time (product_id, create_time DESC);
效果数据：单次请求数据库查询耗时从520ms降到90ms，接口总耗时降到280ms。
踩坑：selectinload默认是lazy='select'，必须显式指定selectinload，否则SQLAlchemy 1.4会退化成N+1。另外注意limit子查询的写法——如果你想要每个商品只取3条评价，需要用lateral join或者应用层分组，我这里是直接用selectinload加载所有评价后取前3，商品数量不多（100以内）的情况下没问题。
五、迁移到FastAPI + 异步ORM
Flask的同步模型在IO密集型场景下性能一般。我决定迁移到FastAPI，利用asyncio和async SQLAlchemy来提升并发能力。
方案设计：用FastAPI替换Flask路由，SQLAlchemy 2.0的异步模式，配合async def处理请求。
# main_fastapi.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import select
from sqlalchemy.orm import selectinload
import aioredis
DATABASE_URL = "mysql+aiomysql://user:pass@localhost/db?charset=utf8mb4"
engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10, echo=False)
async_session = async_sessionmaker(engine, expire_on_commit=False)
class ProductService:
@staticmethod
async def get_product_list(db: AsyncSession):
stmt = (
select(Product)
.options(selectinload(Product.stock))
.options(selectinload(Product.reviews))
.where(Product.status == 1)
.order_by(Product.create_time.desc())
.limit(100)
)
result = await db.execute(stmt)
products = result.scalars().all()
# 手动序列化以避免SQLAlchemy对象延迟加载
data = []
for p in products:
reviews = sorted(p.reviews, key=lambda r: r.create_time, reverse=True)[:3]
data.append({
"id": p.id,
"name": p.name,
"stock": p.stock.quantity if p.stock else 0,
"reviews": [
{"content": r.content, "create_time": r.create_time.isoformat()}
for r in reviews
]
})
return data
app = FastAPI()
@app.get("/products")
async def get_products(db: AsyncSession = Depends(get_db)):
data = await ProductService.get_product_list(db)
return {"code": 0, "data": data}
核心优化点：
1. 异步数据库查询：await db.execute()非阻塞
2. 手动序列化：避免FastAPI自动序列化时触发ORM延迟加载
3. 连接池配置：pool_size=20, max_overflow=10匹配MySQL最大连接数
踩坑：SQLAlchemy 2.0异步模式下，scalars().all()会返回AsyncResult，需要await。另外，expire_on_commit=False必须加，否则事务结束后对象会过期，在序列化时再次查询数据库。
六、缓存策略：Redis缓存热点数据
数据库查询优化后，接口耗时降到120ms，但还没达到目标。考虑加入Redis缓存，缓存商品列表和评价数据，设置合理的过期时间。
方案设计：缓存key设计为products:list:page:1，TTL 60秒。使用Redis的String类型存储JSON序列化后的数据，配合aioredis异步客户端。
# cache_layer.py
import json
from aioredis import Redis
from typing import Optional
CACHE_TTL = 60 # 秒
CACHE_PREFIX = "products:"
class ProductCache:
@staticmethod
def build_cache_key(page: int = 1) -> str:
return f"{CACHE_PREFIX}list:page:{page}"
@staticmethod
async def get_from_cache(redis: Redis, key: str) -> Optional[dict]:
data = await redis.get(key)
return json.loads(data) if data else None
@staticmethod
async def set_to_cache(redis: Redis, key: str, data: dict, ttl: int = CACHE_TTL):
await redis.setex(key, ttl, json.dumps(data, default=str))
# 在接口中使用
@app.get("/products")
async def get_products(
redis: Redis = Depends(get_redis),
db: AsyncSession = Depends(get_db)
):
cache_key = ProductCache.build_cache_key()
cached = await ProductCache.get_from_cache(redis, cache_key)
if cached:
return cached
data = await ProductService.get_product_list(db)
await ProductCache.set_to_cache(redis, cache_key, data)
return {"code": 0, "data": data}
优化细节：
- 缓存过期时间设为60秒，业务允许最多1分钟的数据延迟
- 用json.dumps(data, default=str)处理datetime序列化
- 防止缓存雪崩：在批量更新商品时，主动删除缓存key，而不是等待过期
效果数据：加了缓存后，缓存命中时接口耗时45ms，QPS从120飙升到2100。
七、压测数据对比：最终效果
使用wrk对三个版本进行压测，结果如下：
| 版本 | P50延迟 | P99延迟 | QPS | 错误率 |
|---|---|---|---|---|
| Flask原始版 | 520ms | 800ms | 120 | 0.5% |
| 优化数据库查询后 | 180ms | 280ms | 450 | 0.0% |
| FastAPI异步版 | 80ms | 120ms | 1100 | 0.0% |
| FastAPI+Redis缓存 | 25ms | 45ms | 2100 | 0.0% |
完整压测输出示例（FastAPI+Redis）：
Running 30s test @ http://localhost:8000/products
4 threads and 32 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 23.45ms 12.34ms 180.56ms 81.23%
Req/Sec 542.12 45.23 650.00 72.50%
Latency Distribution
50% 21.00ms
75% 28.00ms
90% 36.00ms
99% 45.00ms
16234 requests in 30.00s, 3.12GB read
Requests/sec: 2165.14
总结：API性能调优的三个核心步骤：
1. Profiling先行：不要凭感觉优化，用数据说话。cProfile或py-spy都可以快速定位热点。
2. 数据库是最大瓶颈：N+1查询、缺少索引、重复序列化是常见问题。ORM加载策略要仔细配置。
3. 异步+缓存是放大器：FastAPI的异步模型能提升并发处理能力，Redis缓存则直接消除数据库压力。
最后给个建议：调优前先设一个明确的目标（比如P99<50ms），然后按步骤执行，每一步都压测量化，不要跳步。如果数据没达到预期，说明还有没发现的瓶颈，继续profiling。
