一、问题背景：扩容救不了的接口
去年接手一个内部商品聚合服务，对外提供 /api/v1/products/search 接口，日均调用量约120万，峰值QPS在400左右。这个接口在FastAPI（0.110.0）里实现，旁边还有个历史遗留的Flask（3.0.2）管理后台接口共用同一套数据库访问层。
问题表现很典型：P95响应时间稳定在1800ms上下，P99能到3.2s。运维第一反应是扩容，从2个Pod加到8个，CPU利用率反而更低了，P95只降到1650ms。这说明瓶颈根本不在计算资源上，而是每个请求都在等某个东西。
更麻烦的是慢查询日志里看不到明显异常——数据库侧的慢查询阈值设的1s，几乎没触发。也就是说时间花在了"很多次不算慢的小查询"上，这种问题最容易被忽略。
目标很明确：P95降到200ms以内，QPS至少翻三倍，且不增加Pod数量。
二、环境与版本
先把环境交代清楚，后面所有数字都基于这套配置：
- Python 3.11.6
- FastAPI 0.110.0 + Uvicorn 0.27.1（workers=4，单Pod 4核）
- Flask 3.0.2（gunicorn 21.2.0，sync worker）
- SQLAlchemy 2.0.27，PostgreSQL 15.6
- Redis 7.2.4，redis-py 5.0.1
- 压测工具：locust 2.24.0，压测机 8核16G，与被压服务同内网
数据库连接池用的是SQLAlchemy默认的QueuePool，pool_size=5，max_overflow=10——这个默认值后面会成为坑点之一。
三、方案设计：先定位，再动手
我的原则是先profiling再改代码，绝不凭直觉优化。三步走：
第一步，用py-spy做线上采样。它是采样型profiler，对运行中的进程attach，开销极低，适合生产环境：
py-spy record -o profile.svg --pid --duration 60 --rate 200
同时用cProfile在预发环境跑单请求级别的火焰图，配合flask侧的--profile模式对比两个框架的表现。结论很有意思：FastAPI和Flask的框架开销占比都不到8%，时间几乎全部落在sqlalchemy.orm.session的execute调用栈里，而且调用次数惊人——单个请求平均触发47次SQL。
第二步，打印SQL计数。写个中间件统计每请求查询数，确认是N+1问题：
# middleware.py —— 挂在FastAPI上统计单请求SQL次数
import time
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy import event
from app.db import engine
class SQLCountMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
counter = {"n": 0, "t": 0.0}
def before_execute(conn, cursor, statement, params, context, executemany):
counter["n"] += 1
def after_cursor_execute(conn, cursor, statement, params, context, executemany):
pass
event.listen(engine, "before_cursor_execute", before_execute)
start = time.perf_counter()
response = await call_next(request)
cost = (time.perf_counter() - start) * 1000
event.remove(engine, "before_cursor_execute", before_execute)
response.headers["X-SQL-Count"] = str(counter["n"])
response.headers["X-Cost-Ms"] = f"{cost:.1f}"
return response
结果很直白：搜索接口平均47条SQL，其中商品主表1条，SKU表1条，剩下45条全是循环里查分类名、品牌名、库存——典型N+1。
第三步，确定优化组合：批量查询消除N+1、Redis缓存热点数据、调整连接池。缓存策略上我选了两级——进程内LRU（分类/品牌这类几乎不变的字典型数据）+ Redis（商品详情，TTL 60s + 主动失效）。
四、核心实现
4.1 消除N+1：从47条SQL到3条
原代码大致长这样（简化）：
# 优化前：循环内查询，47条SQL的元凶
def search_products(keyword: str, limit: int = 20):
rows = db.execute(text(
"SELECT id, name, brand_id, category_id FROM products "
"WHERE name ILIKE :kw LIMIT :lim"
), {"kw": f"%{keyword}%", "lim": limit}).fetchall()
result = []
for r in rows:
brand = db.execute(text("SELECT name FROM brands WHERE id=:id"),
{"id": r.brand_id}).scalar()
category = db.execute(text("SELECT name FROM categories WHERE id=:id"),
{"id": r.category_id}).scalar()
stock = db.execute(text("SELECT qty FROM stocks WHERE product_id=:id"),
{"id": r.id}).scalar()
result.append({**dict(r._mapping), "brand": brand,
"category": category, "stock": stock})
return result
改写思路：用IN批量取关联数据，再用字典做内存join。注意ILIKE '%kw%'走不了B-tree索引，我额外加了pg_trgm的GIN索引（这个后面单独说）：
# 优化后：固定3条SQL，与结果条数无关
from sqlalchemy import text
def search_products(db, keyword: str, limit: int = 20):
rows = db.execute(text(
"SELECT id, name, brand_id, category_id FROM products "
"WHERE name ILIKE :kw ORDER BY id LIMIT :lim"
), {"kw": f"%{keyword}%", "lim": limit}).mappings().all()
if not rows:
return []
pids = [r["id"] for r in rows]
bids = {r["brand_id"] for r in rows}
cids = {r["category_id"] for r in rows}
# 三条批量查询，用 IN 一次取回
brands = dict(db.execute(text(
"SELECT id, name FROM brands WHERE id = ANY(:ids)"),
{"ids": list(bids)}).all())
categories = dict(db.execute(text(
"SELECT id, name FROM categories WHERE id = ANY(:ids)"),
{"ids": list(cids)}).all())
stocks = dict(db.execute(text(
"SELECT product_id, qty FROM stocks WHERE product_id = ANY(:ids)"),
{"ids": pids}).all())
return [{**dict(r),
"brand": brands.get(r["brand_id"]),
"category": categories.get(r["category_id"]),
"stock": stocks.get(r["id"], 0)} for r in rows]
单这一项，P95从1800ms降到约520ms——因为省掉了45次网络往返，每次往返哪怕只有8ms，累计也是360ms起步，加上连接池争用后放大得更厉害。
4.2 两级缓存
分类、品牌这类数据变更频率极低（一天几次），用cachetools的TTL LRU放进程内，命中率接近100%；商品详情用Redis，60秒TTL，后台更新时主动DEL。
# cache.py
import json
from cachetools import TTLCache
from redis import Redis
local_cache = TTLCache(maxsize=2048, ttl=300) # 进程内，5分钟
redis = Redis(host="redis.internal", port=6379, db=0,
socket_timeout=0.05, socket_connect_timeout=0.05,
max_connections=64, health_check_interval=30)
def get_product(pid: int):
if pid in local_cache:
return local_cache[pid]
raw = redis.get(f"p:{pid}")
if raw:
data = json.loads(raw)
local_cache[pid] = data
return data
data = _load_from_db(pid) # 回源
redis.setex(f"p:{pid}", 60, json.dumps(data, default=str))
local_cache[pid] = data
return data
注意socket_timeout=0.05——缓存是加速手段，绝不能成为新的故障点。50ms超时后直接回源，宁可慢一点也不能挂。
4.3 连接池参数
默认pool_size=5, max_overflow=10在4个uvicorn worker下就是20个连接上限，而压测峰值并发约120，大量请求卡在等连接。调成：
engine = create_engine(
DATABASE_URL,
pool_size=20, max_overflow=30, # 单worker最多50连接
pool_pre_ping=True,
pool_recycle=1800,
pool_timeout=2, # 等连接超时2s，快速失败
echo=False,
)
PostgreSQL侧max_connections设的300，4 worker × 50 = 200，留了余量。
五、踩坑与优化
坑1：Redis大key。 最初把整个商品列表（含详情）塞进一个key，单key 2MB，网络传输反而成了瓶颈。改成按商品ID缓存单条，列表只缓存ID数组。
坑2：GIN索引没生效。 加了pg_trgm后ILIKE '%kw%'还是全表扫描，原因是索引建在lower(name)上而查询没走表达式。要么改查询为lower(name) ILIKE lower(:kw)，要么建在原始列上。我选了后者，简单直接。另外中文分词场景pg_trgm效果一般，数据量再大就得考虑ES了，但当前20万行数据GIN够用。
坑3：Flask侧的同步阻塞。 管理后台的Flask接口用gunicorn sync worker，压测时把数据库连接池占满，影响了FastAPI侧。给Flask单独配了连接池（pool_size=5），并给两个服务做了数据库账号隔离，避免互相拖累。Flask这边顺手把SQLALCHEMY_ENGINE_OPTIONS里的pool_pre_ping打开，解决了偶发的"server closed the connection unexpectedly"。
坑4：压测方法本身。 一开始locust只压了单接口，结果很好看，上线后发现真实流量里搜索和详情混跑，指标差一截。后来用加权任务模拟真实比例（搜索:详情=3:7）重新压测，数据才可信。
六、效果数据
压测条件统一：locust 200并发，持续10分钟，搜索:详情=3:7，8个Pod不变。
| 指标 | 优化前 | 优化后 | 变化 |
|---|---|---|---|
| P50 | 920ms | 38ms | -96% |
| P95 | 1800ms | 95ms | -95% |
| P99 | 3200ms | 210ms | -93% |
| QPS（单Pod） | 26 | 300 | +1054% |
| 单请求平均SQL数 | 47 | 3 | -94% |
| Redis命中率 | — | 91.3% | — |
| 错误率 | 0.8%（连接超时） | 0.02% | — |
上线后观察一周，P95稳定在90-110ms区间，数据库CPU从峰值78%降到21%，最直接的好处是省了原计划的4个Pod扩容预算。
七、总结
复盘下来有三点值得记住：
一是profiling的价值远大于经验直觉。我一开始也怀疑是FastAPI的async开销或者Pydantic序列化慢，py-spy一采样才发现框架层只占8%，47次SQL才是真凶。没有数据支撑的优化基本是碰运气。
二是N+1是API性能的头号杀手，且它不产生慢查询。每条SQL都只要几毫秒，慢查询日志完全看不到，但累计起来就是灾难。建议所有团队都在中间件里加个每请求SQL计数，超过阈值告警。
三是缓存要设超时和降级。Redis的50ms socket_timeout和回源逻辑救过一次Redis抖动，否则缓存层故障会直接变成雪崩。
后续还打算做两件事：把搜索换成tsvector全文索引（中文场景pg_trgm的召回率还是偏弱），以及给热点商品加本地二级缓存预热。有进展再写一篇。
代码都在上面了，环境版本也标清楚了，拿去能直接跑。如果你的接口也在"扩容无效"的阶段，先别加机器，挂个py-spy看看时间到底花在哪。
