在当今快节奏的Web应用开发中，API性能直接影响用户体验和系统稳定性。FastAPI作为一个高性能的Python Web框架，凭借异步支持和自动生成文档等特性，已成为众多项目的首选。但即使使用FastAPI，若代码和部署环节缺乏优化，性能瓶颈仍可能出现。本文将从代码、数据库、缓存到部署，一步步拆解FastAPI性能优化的核心思路和实用技巧，帮助初学者快速提升应用效率。
一、代码层面的基础优化¶
FastAPI本身已基于异步和高效解析器（如Uvicorn）实现了基础性能优势，但代码写法仍可能成为瓶颈。以下是初学者最容易优化的方向：
1. 优先使用异步函数处理IO密集型任务¶
FastAPI支持async def定义异步路径操作函数，适合处理IO密集型任务（如数据库查询、API调用、文件读写）。但需注意：不要在异步函数中执行CPU密集型操作（如复杂循环、数学计算），这类任务会阻塞事件循环，反而降低性能。
# 反例：同步数据库查询（会阻塞事件循环）
def get_user_sync(user_id: int):
db = create_db_connection() # 同步连接
result = db.query("SELECT * FROM users WHERE id = ?", user_id)
db.close()
return result
# 正例：异步数据库查询（非阻塞）
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine # 异步ORM
app = FastAPI()
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") # 异步连接
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(User.__table__.select().where(User.id == user_id)) # 用await等待异步查询
return result.scalars().first()
2. 避免“过度计算”和数据冗余¶
在路径函数中，若需返回大量数据（如列表），直接构造完整列表可能导致内存占用过高。此时可用生成器（Generator） 或分页替代，减少内存消耗。
# 反例：返回完整大列表（内存占用高）
@app.get("/products")
async def get_all_products():
return [{"id": i, "name": f"商品{i}"} for i in range(10000)] # 生成10000条数据
# 正例：使用生成器+分页（按需返回）
from fastapi import Query
@app.get("/products")
async def get_products(page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100)):
offset = (page - 1) * page_size
# 生成器示例：通过SQL的limit/offset分页，只返回当前页数据
async with engine.connect() as conn:
result = await conn.execute(
User.__table__.select().offset(offset).limit(page_size)
)
return [dict(row) for row in result.all()]
3. 用参数验证减少后续处理¶
FastAPI的参数验证（如类型注解、Query/Path参数）不仅能自动生成API文档，还能提前过滤无效请求，避免后续逻辑浪费资源。
# 正例：参数类型注解+范围验证
@app.get("/items/{item_id}")
async def get_item(item_id: int = Path(..., ge=1), limit: int = Query(10, le=100)): # 限制item_id≥1，limit≤100
return {"item_id": item_id, "limit": limit}
二、异步编程的正确姿势¶
异步是FastAPI的核心优势，但误用异步反而会导致性能下降。以下是关键原则：
1. 区分“IO密集型”和“CPU密集型”任务¶
- IO密集型（如网络请求、数据库查询）：适合异步，因为等待IO时可切换其他任务。
- CPU密集型（如大数据计算、AI推理）：异步不适用，需用
asyncio.run_in_executor提交到线程池，或改用多进程。
# 反例：异步函数中执行CPU密集任务（阻塞事件循环）
import time
async def process_data(data: list):
result = []
for item in data:
time.sleep(0.1) # 模拟CPU密集计算（实际中可能是for循环处理大数据）
result.append(item * 2)
return result
# 正例：将CPU密集任务丢到线程池（非阻塞）
from concurrent.futures import ThreadPoolExecutor
@app.get("/process")
async def process_data():
data = [1, 2, 3, ..., 1000] # 大数据列表
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
ThreadPoolExecutor(), # 线程池处理CPU密集任务
process_data_sync, # 同步函数（内部是CPU密集逻辑）
data
)
return result
2. 异步框架的正确搭配¶
使用异步ORM（如SQLAlchemy 1.4+的异步版本、Tortoise-ORM）或异步HTTP客户端（如aiohttp），避免在异步函数中调用同步库（如requests）。
# 错误：异步函数中调用同步HTTP库（会阻塞事件循环）
import requests
async def get_remote_data():
response = requests.get("https://api.example.com/data") # 同步请求，会阻塞！
return response.json()
# 正确：用异步HTTP客户端
import aiohttp
async def get_remote_data():
async with aiohttp.ClientSession() as session:
async with session.get("https://api.example.com/data") as response:
return await response.json() # 等待异步响应
三、数据库查询优化¶
数据库是最常见的性能瓶颈，优化方向如下：
1. 连接池与连接复用¶
频繁创建/关闭数据库连接会消耗大量时间。使用连接池复用连接，设置合理的连接数（通常为CPU核心数×2）。
# SQLAlchemy异步连接池配置
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from sqlalchemy.orm import sessionmaker
engine = AsyncEngine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=10, # 连接池大小（根据并发量调整）
max_overflow=20, # 超过pool_size后最多创建的连接数
pool_recycle=300 # 连接超时自动回收（避免数据库断连）
)
SessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
2. 索引与查询语句优化¶
- 加索引：确保查询字段（如
WHERE、JOIN条件）有索引，避免全表扫描。 - 优化SQL：避免
SELECT *（只查必要字段），用LIMIT/OFFSET分页（大数据量用游标分页）。
-- 优化前：无索引的全表扫描
SELECT * FROM users WHERE name LIKE '%test%'; # 慢！全表扫描
-- 优化后：加索引
CREATE INDEX idx_users_name ON users(name); # 对name字段加索引
SELECT id, email FROM users WHERE name LIKE '%test%' LIMIT 100; # 只查必要字段+分页
3. 延迟加载与批量操作¶
- 延迟加载：用
selectinload（批量加载关联数据）替代joinedload（立即加载），减少数据库往返。 - 批量插入/更新：用
bulk_insert_mappings替代逐条插入，减少SQL执行次数。
四、缓存策略：减少重复计算¶
缓存可显著降低数据库负载和重复计算，适合不常变化、高频访问的数据（如热门商品列表、配置信息）。
1. 内存缓存（简单场景）¶
用cachetools实现内存缓存，适合单实例部署：
from cachetools import LRUCache
# 定义缓存（最多存100条记录，自动淘汰旧数据）
cache = LRUCache(maxsize=100)
@app.get("/hot-products")
async def get_hot_products():
# 尝试从缓存获取
cached_result = cache.get("hot_products")
if cached_result:
return cached_result # 直接返回缓存数据
# 缓存未命中，查询数据库
async with SessionLocal() as session:
result = await session.execute("SELECT * FROM products ORDER BY sales DESC LIMIT 20")
hot_products = [dict(row) for row in result.all()]
# 存入缓存（设置10分钟过期）
cache["hot_products"] = hot_products
return hot_products
2. Redis分布式缓存（多实例场景）¶
多服务器部署时，用Redis作为共享缓存，支持跨实例缓存。需安装redis库：
import redis
r = redis.Redis(host="localhost", port=6379, db=0) # 连接Redis
@app.get("/product/{product_id}")
async def get_product(product_id: int):
cache_key = f"product:{product_id}"
cached_data = r.get(cache_key)
if cached_data:
return json.loads(cached_data) # 返回缓存数据
# 数据库查询
product = await get_product_from_db(product_id)
# 存入Redis（设置30分钟过期）
r.setex(cache_key, 60*30, json.dumps(product))
return product
五、部署与扩展：提升并发能力¶
代码和数据库优化后，需通过部署配置提升服务器并发处理能力。
1. 多进程/多线程部署¶
FastAPI需配合异步服务器（如Uvicorn）和进程管理工具（如Gunicorn）：
# 启动命令：Uvicorn + Gunicorn（2个worker，每个worker4个线程）
gunicorn main:app -w 2 -k uvicorn.workers.UvicornWorker -t 120 --threads 4
- worker数量：通常为
CPU核心数×2 + 1（充分利用多核）。 - 线程数：IO密集型任务设为
worker数×CPU核心数，减少等待时间。
2. 反向代理与负载均衡¶
用Nginx作为反向代理，处理静态资源、SSL终结，并转发请求到多个FastAPI实例：
# Nginx配置示例
server {
listen 80;
location / {
proxy_pass http://127.0.0.1:8000; # 转发到FastAPI
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
location /static/ {
alias /path/to/static/files/; # 直接返回静态资源，不经过FastAPI
}
}
3. 容器化与自动扩缩容¶
用Docker容器化部署，配合Kubernetes实现自动扩缩容（根据CPU/内存使用率动态增加实例）：
# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]
总结：性能优化的核心思路¶
- 定位瓶颈：用工具（如
cProfile、asyncio调试）先找出慢的代码段。 - 从代码到部署逐步优化：先优化异步代码→数据库→缓存→部署。
- 优先解决高性价比问题：比如加索引、缓存，往往比复杂架构更简单有效。
- 监控与迭代：用Prometheus、APM工具监控性能指标，持续优化。
通过以上步骤，即使是初学者也能系统提升FastAPI的性能，打造高效、稳定的API服务。