一、性能优化概述

1.1 常见瓶颈

• 数据库查询慢

• 同步I/O阻塞

• 不必要的序列化

• 缺乏缓存

1.2 优化维度

• 异步处理

• 数据库优化

• 缓存策略

• 服务器配置

二、异步处理

2.1 正确使用async

# 正确：异步数据库
async def get_user_async(user_id: int):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

# 正确：异步HTTP调用
async def fetch_external_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

2.2 避免阻塞

# 错误：同步调用
def sync_operation():
    time.sleep(1)  # 阻塞整个事件循环

# 正确：异步调用
async def async_operation():
    await asyncio.sleep(1)  # 非阻塞

三、数据库优化

3.1 连接池

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
)

3.2 查询优化

# 使用selectin加载避免N+1
result = await db.execute(
    select(User).options(selectinload(User.orders))
)

四、缓存策略

4.1 内存缓存

from cachetools import TTLCache

cache = TTLCache(maxsize=1000, ttl=300)

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id in cache:
        return cache[item_id]

    item = await db.get(Item, item_id)
    cache[item_id] = item
    return item

4.2 Redis缓存

import redis.asyncio as redis

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    cached = await redis.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    user = await db.get(User, user_id)
    await redis.setex(f"user:{user_id}", 300, json.dumps(user))
    return user

五、部署配置

5.1 Uvicorn优化

# 生产环境配置
uvicorn main:app     --workers 4     --host 0.0.0.0     --port 8000     --limit-concurrency 1000     --limit-max-requests 10000

5.2 Gunicorn配置

# gunicorn_config.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
max_requests = 10000
max_requests_jitter = 1000

六、总结

FastAPI性能优化要点：
1. 全面异步：避免阻塞
2. 连接池：复用数据库连接
3. 合理缓存：减少重复计算
4. 正确配置：充分利用硬件