FastAPI异步改造与Redis缓存：记一次API响应从2.1s到180ms的调优

Python 3.11.4
FastAPI 0.104.1
uvicorn 0.24.0 (worker=4, 每个worker异步模式)
SQLAlchemy 2.0.23 (asyncpg驱动)
PostgreSQL 14.5 (云RDS, 4核8G)
Redis 7.0.12 (单机, 最大内存2G)
Docker 24.0.5

# 进入容器，找到uvicorn主进程PID
kubectl exec -it my-api-pod-xxx -- sh
# 安装py-spy（生产环境临时用，用完就卸）
pip install py-spy==0.3.14
# 抓取20秒的调用栈，输出到文件
py-spy record -o /tmp/profile.svg -p $(pgrep -f uvicorn | head -1) --duration 20
# 或者直接dump当前栈
py-spy dump --pid $(pgrep -f uvicorn | head -1)

# 优化前：同步requests阻塞事件循环
@app.get("/api/v1/report/summary")
async def get_summary(request: Request):
    # 同步调用外部服务，阻塞事件循环！
    resp = requests.get("http://auth-service/api/check", timeout=0.5)
    if resp.status_code != 200:
        raise HTTPException(status_code=403)
    # 查询数据库
    data = await db.query(ReportSummary).filter(...).all()
    return {"data": data}

# 优化后：使用httpx异步客户端
import httpx
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_http_client():
    async with httpx.AsyncClient(timeout=0.5) as client:
        yield client

@app.get("/api/v1/report/summary")
async def get_summary(request: Request):
    # 异步调用外部服务，不阻塞事件循环
    async with get_http_client() as client:
        resp = await client.get("http://auth-service/api/check")
        if resp.status_code != 200:
            raise HTTPException(status_code=403)
    # 查询数据库（异步）
    data = await db.query(ReportSummary).filter(...).all()
    return {"data": data}

# 优化前：懒加载导致N+1
stmt = select(Report).where(Report.tenant_id == tenant_id)
reports = await db.execute(stmt)
for report in reports.scalars().all():
    # 每次访问report.items都会触发一次数据库查询
    print(len(report.items))

# 优化后：使用selectinload预加载关联表
from sqlalchemy.orm import selectinload

stmt = (
    select(Report)
    .options(selectinload(Report.items))  # 一条IN查询替代N条SELECT
    .where(Report.tenant_id == tenant_id)
)
reports = await db.execute(stmt)
for report in reports.scalars().all():
    print(len(report.items))

from cachetools import TTLCache
import redis.asyncio as aioredis

# 本地缓存，每个worker独立
local_cache = TTLCache(maxsize=128, ttl=60)
# Redis连接池
redis_pool = aioredis.ConnectionPool.from_url(
    "redis://localhost:6379/0", max_connections=20
)
redis_client = aioredis.Redis(connection_pool=redis_pool)

async def get_report_data(tenant_id: str):
    # 1. 查本地缓存
    cache_key = f"report_summary:{tenant_id}"
    if cache_key in local_cache:
        return local_cache[cache_key]

    # 2. 查Redis（加锁避免击穿）
    async with redis_client.lock(f"lock:{cache_key}", timeout=5):
        # 双重检查：锁获取后再查一次Redis
        cached = await redis_client.get(cache_key)
        if cached:
            data = json.loads(cached)
            local_cache[cache_key] = data  # 回填本地
            return data

        # 3. 查数据库
        data = await query_db(tenant_id)
        # 4. 回填缓存
        await redis_client.set(cache_key, json.dumps(data), ex=300)
        local_cache[cache_key] = data
        return data

# 模块级复用
_client = None

async def get_client():
    global _client
    if _client is None:
        _client = httpx.AsyncClient(timeout=0.5, limits=httpx.Limits(max_connections=100))
    return _client

try:
    async with redis_client.lock(...):
        ...
except Exception as e:
    # Redis不可用，直接查数据库
    logger.error(f"Redis lock failed: {e}")
    return await query_db(tenant_id)

wrk -t 12 -c 300 -d 30s --latency http://api.example.com/api/v1/report/summary