一、问题背景：一个“简单”的订单接口为何拖垮数据库

上周五下午，运营反馈后台订单列表页转圈超过3秒。我看了一下监控，/api/v1/orders 接口P95耗时458ms，高峰期直接飙到912ms，数据库CPU使用率92%。这个接口逻辑看起来很简单：查订单表、关联用户表、关联商品表，然后返回列表。但就是这样一个“简单”接口，成了整个系统的瓶颈。

我的第一反应是看慢查询日志，结果发现并没有特别慢的SQL——单条查询都在50ms以内。那就不是SQL本身的问题，而是查询次数太多了。

二、环境与版本：技术栈全景

先交代一下生产环境：

  • Python 3.10.12
  • FastAPI 0.104.1 + Uvicorn 0.24.0（workers=4）
  • SQLAlchemy 2.0.23（async模式）+ asyncpg 0.29.0
  • PostgreSQL 14.5（16核32G）
  • Redis 7.0.12（单节点）
  • 压测工具：wrk 4.2.0 + 自建脚本

服务部署在K8s，4个Pod副本，每个Pod限制2核4G。数据库是托管实例，连接池上限50。

三、方案设计：先用工具定位，别瞎猜

3.1 用py-spy做生产环境采样分析

慢查询日志没问题，那就得看应用层的调用堆栈。我用py-spy对生产Pod做了60秒采样：

# 找到目标Pod的PID
kubectl exec -it orders-api-7d8f9f6d8c-abcde -- py-spy dump --pid 1

采样结果让我很惊讶：OrderService.get_order_list 方法占了78%的CPU时间，其中大部分花在了 await session.execute() 上。这说明接口在循环里执行了多次查询——典型的N+1问题。

3.2 用cProfile分析单请求调用链

本地复现后，用cProfile跑了一次请求：

import cProfile
import pstats
from app.main import app
from httpx import AsyncClient

async def bench():
    async with AsyncClient(app=app, base_url="http://test") as client:
        for _ in range(50):
            await client.get("/api/v1/orders?page=1&page_size=20")

profiler = cProfile.Profile()
profiler.enable()
import asyncio
asyncio.run(bench())
profiler.disable()
stats = pstats.Stats(profiler).sort_stats("cumulative")
stats.print_stats(30)

关键输出：

ncalls  tottime  percall  cumtime  percall  filename:lineno(function)
   50    0.002    0.000    0.412    0.008  orders.py:56(get_order_list)
  500    0.001    0.000    0.356    0.001  orders.py:87(get_user_by_id)
  500    0.001    0.000    0.298    0.001  orders.py:102(get_product_by_id)

看到了吧：50次请求，500次用户查询、500次商品查询。每个订单都单独查一次用户和商品，这就是问题的根源。

四、核心实现：三步优化

4.1 第一步：SQLAlchemy联合查询消除N+1

原来的代码是循环里查关联表，我改成了一次性join：

# 优化前：循环查询
async def get_order_list_before(db: AsyncSession, page: int, page_size: int):
    result = await db.execute(
        select(Order).order_by(Order.created_at.desc())
        .offset((page - 1) * page_size).limit(page_size)
    )
    orders = result.scalars().all()
    items = []
    for order in orders:  # 每次循环2次查询，20条订单就是40次
        user = await db.get(User, order.user_id)
        product = await db.get(Product, order.product_id)
        items.append({
            "order_id": order.id,
            "user_name": user.name,
            "product_name": product.name,
            "amount": order.amount,
        })
    return items

# 优化后：一次性join
async def get_order_list_after(db: AsyncSession, page: int, page_size: int):
    stmt = (
        select(Order, User, Product)
        .join(User, Order.user_id == User.id)
        .join(Product, Order.product_id == Product.id)
        .order_by(Order.created_at.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
    )
    result = await db.execute(stmt)
    rows = result.all()  # 一次查询搞定
    return [
        {
            "order_id": order.id,
            "user_name": user.name,
            "product_name": product.name,
            "amount": order.amount,
        }
        for order, user, product in rows
    ]

这一步改动后，数据库查询次数从41次降到1次。本地压测P95从458ms降到了112ms，效果立竿见影。

4.2 第二步：Redis缓存热点数据

查询次数少了，但用户表和商品表的数据其实是相对稳定的。我加了一层Redis缓存，TTL设为10分钟：

import json
import redis.asyncio as aioredis

redis_client = aioredis.from_url(
    "redis://redis-service:6379/0",
    max_connections=20,
    decode_responses=True
)

async def get_user_cached(user_id: int) -> dict | None:
    cache_key = f"user:info:{user_id}"
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    # 从数据库查询（这里省略，假设用SQLAlchemy）
    user = await fetch_user_from_db(user_id)
    if user:
        await redis_client.set(cache_key, json.dumps(user), ex=600)  # 10分钟过期
    return user

注意：缓存只加在用户和商品信息上，订单数据没有缓存，因为订单实时性要求高，而且量大不适合全量缓存。

4.3 第三步：引入lru_cache做进程内缓存

Redis缓存减少了跨网络调用，但每次都走一次Redis协议也有开销。对于用户信息这种几乎不变的场景，我直接用Python内置的functools.lru_cache做了进程内缓存：

from functools import lru_cache

@lru_cache(maxsize=1024)
def get_user_info_sync(user_id: int) -> dict:
    # 注意：这是同步函数，实际使用时要结合线程池
    # 或者用async_lru库
    return {
        "id": user_id,
        "name": f"user_{user_id}",
    }

# 对于异步场景，我用了async_lru库
from async_lru import alru_cache

@alru_cache(maxsize=1024, ttl=300)  # 5分钟过期
async def get_user_info_async(user_id: int) -> dict:
    # 先查Redis，再查DB
    user = await get_user_cached(user_id)
    return user

这里踩了一个坑：lru_cache不能直接用于async函数，会报TypeError: the wrapped function is no longer a coroutine。必须用async_lru库或者自己封装。

五、踩坑与优化：版本兼容和缓存一致性

5.1 坑一：asyncpg与SQLAlchemy2.0的兼容问题

升级SQLAlchemy到2.0后，原来用的await session.query()全部废弃了，必须改成select()语法。这个迁移花了我半天时间，主要是各种关联查询的写法变了。

5.2 坑二：缓存雪崩和击穿

加Redis缓存后，遇到一个问题：用户信息缓存在同一时间过期，导致大量请求同时穿透到数据库。解决方案是给TTL加随机抖动：

import random
ttl = 600 + random.randint(0, 120)  # 10分钟±2分钟随机
await redis_client.set(cache_key, json.dumps(user), ex=ttl)

5.3 坑三：Uvicorn workers与进程内缓存

lru_cache是进程级别的，4个worker意味着每个进程都有独立的缓存。这会导致缓存不一致，但用户信息这种场景影响不大。如果需要严格一致，可以用cachetools配合Redis做分布式锁。

六、压测数据：优化前后对比

最终用wrk做了压测，命令如下：

wrk -t8 -c100 -d30s --latency http://localhost:8000/api/v1/orders?page=1&page_size=20

结果对比：

指标 优化前 优化后 提升
P50延迟 312ms 28ms 11.1x
P95延迟 458ms 39ms 11.7x
P99延迟 912ms 87ms 10.5x
吞吐量 320 req/s 2100 req/s 6.6x
DB CPU 92% 23% -75%
每请求DB查询次数 41 1 41x

从数据看，优化效果非常显著。核心收益来自两点：联合查询消除了N+1，缓存消除了重复计算和网络IO。

七、总结：性能调优的黄金法则

这次调优让我深刻体会到几件事：

  1. 先profile再优化，不要凭感觉猜瓶颈。py-spy和cProfile让我十分钟就定位到了N+1问题。
  2. 数据库查询优化是第一步，缓存是第二步。如果SQL本身有问题，加再多缓存也白搭。
  3. 缓存要分层：进程内缓存（LRU） > Redis缓存 > 数据库查询，每一层的延迟差一个数量级。
  4. 注意缓存过期策略，随机TTL能有效防止雪崩。

最后说一句：这个接口的QPS峰值也就2000左右，如果你遇到的是几万QPS的场景，那可能还需要引入消息队列削峰、读写分离等更复杂的架构。但不管多复杂的系统，调优的思路都是一样的：先定位，再优化，用数据说话。

代码仓库：github.com/yourname/orders-api-tuning（示例代码已脱敏）

延伸阅读：如果你对py-spy不熟悉，强烈建议看看它的官方文档。生产环境调试神器，比加日志定位问题快十倍。