FastAPI: 性能优化 — 从百到百万 QPS

Q: run_in_executor 的线程池有多大？

默认 min(32, os.cpu_count() + 4) 。CPU 密集型任务用 ProcessPoolExecutor 避免受 GIL 限制。

Q: orjson 与 Pydantic 兼容吗？

兼容。FastAPI 的 default_response_class=ORJSONResponse 自动用 orjson 序列化 Pydantic 模型的 model_dump() 输出。

最后更新：2026-08-26

性能优化像修高速公路——先修最堵的路段（瓶颈），每修一段测一次效果，而不是到处加宽。盲目优化是浪费，测量驱动的优化才是高效。

1. 你将学到

异步阻塞排查：asyncio 陷阱与 run_in_executor 改造同步代码
数据库优化：索引策略、查询优化、连接池调优
并发模型：Uvicorn workers 数量与 Gunicorn 配置
响应压缩：GZipMiddleware 与 JSON 序列化优化（orjson）
Alice 场景：PriceTracker 从单机 500 QPS 优化到集群 100 万 QPS 的调优路径

2. Alice 的真实故事

(1) 痛点：单机 500 QPS 不够用

PriceTracker 上线后单机只能处理 500 QPS，大促期间流量飙到 5000 QPS，API 响应超时。Charlie 加了服务器但效果不明显——数据库连接池耗尽、同步代码阻塞事件循环、JSON 序列化占用 40% CPU。

(2) 系统化性能优化的解法

性能优化不是盲猜，而是测量→定位瓶颈→优化→验证的循环。从代码级（async/序列化）到数据库级（索引/连接池）再到架构级（Workers/负载均衡），逐层突破。

(3) 收益

PriceTracker QPS 从 500 优化到 10000（单机），加上负载均衡集群可达 100 万 QPS，P99 延迟从 500ms 降到 30ms。

3. 性能优化层级

(1) 三层优化模型

graph TD
    L3[Infrastructure Layer] --> L2[Database Layer]
    L2 --> L1[Application Layer]
    
    L1 --- A1[async/await - Block detection]
    L1 --- A2[Serialization - orjson]
    L1 --- A3[Compression - GZip]
    
    L2 --- D1[Indexes - Query optimization]
    L2 --- D2[Connection Pool - Size tuning]
    L2 --- D3[Query Patterns - N+1 prevention]
    
    L3 --- I1[Workers - Gunicorn config]
    L3 --- I2[Load Balancer - Nginx]
    L3 --- I3[Auto-scaling - K8s HPA]

(2) QPS 提升路径

flowchart LR
    A[500 QPS\nBaseline] -->|async + orjson| B[2000 QPS]
    B -->|DB indexes + pool| C[10000 QPS]
    C -->|4 Workers| D[40000 QPS]
    D -->|Nginx LB × 5| E[200000 QPS]
    E -->|K8s × 5 pods| F[1000000 QPS]

阶段	优化手段	QPS	提升倍数
基线	默认配置	500	1x
应用层	async + orjson	2,000	4x
数据库层	索引 + 连接池	10,000	20x
Workers	4 Workers + Gunicorn	40,000	80x
负载均衡	Nginx × 5 实例	200,000	400x
K8s 弹性	5 Pods 自动伸缩	1,000,000	2000x

4. 应用层优化

(1) 异步阻塞排查

▶ 示例：同步代码阻塞事件循环

PYTHON

import asyncio
import time

# BAD: sync I/O blocks the event loop
async def get_price_bad(product_id: int):
    time.sleep(0.1)  # Blocks ALL other requests for 100ms!
    return {"product_id": product_id, "price": 9.99}

# GOOD: use asyncio for I/O
async def get_price_good(product_id: int):
    await asyncio.sleep(0.1)  # Non-blocking, other requests continue
    return {"product_id": product_id, "price": 9.99}

输出：

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：run_in_executor 包装同步代码

PYTHON

import asyncio
from functools import partial

# Sync function that cannot be made async
def sync_scrape_price(url: str) -> float:
    # Uses requests library (sync only)
    import requests
    response = requests.get(url, timeout=10)
    return parse_price(response.text)

async def get_price_async(product_id: int):
    loop = asyncio.get_event_loop()
    # Run sync function in thread pool - doesn't block event loop
    price = await loop.run_in_executor(
        None,  # Default thread pool
        partial(sync_scrape_price, f"https://api.example.com/prices/{product_id}"),
    )
    return {"product_id": product_id, "price": price}

输出：

TEXT 📖 仅展示

# 函数定义成功

(2) JSON 序列化优化

▶ 示例：orjson 替代 json

PYTHON

# Install: uv add orjson
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

app = FastAPI(default_response_class=ORJSONResponse)

@app.get("/api/v1/products")
async def list_products():
    # orjson is 3-10x faster than stdlib json
    return {"products": [{"id": i, "name": f"Product {i}"} for i in range(100)]}

输出：

TEXT 📖 仅展示

# 函数定义成功

序列化库	速度	说明
json（标准库）	基准	纯 Python 实现
orjson	3-10x	Rust 实现，自动处理 datetime
ujson	2-3x	C 实现，部分兼容性问题

▶ 示例：GZipMiddleware 压缩

PYTHON

from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware

app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)  # Compress responses > 1KB

@app.get("/api/v1/products")
async def list_products():
    # Large JSON responses are automatically compressed
    return {"products": [...]}  # 50KB → ~5KB with gzip

输出：

TEXT 📖 仅展示

# 函数定义成功

5. 数据库层优化

(1) 索引策略

▶ 示例：PriceTracker 索引设计

PYTHON

from sqlalchemy import Index

class Product(Base):
    __tablename__ = "products"
    __table_args__ = (
        # Single-column indexes for common filters
        Index("ix_products_category", "category"),
        Index("ix_products_created_at", "created_at"),
        # Composite index for category + price range queries
        Index("ix_products_category_base_price", "category", "base_price"),
        # Partial index for active products only (PostgreSQL specific)
        Index("ix_products_active", "created_at", postgresql_where=("deleted_at IS NULL")),
    )
    ...

输出：

TEXT 📖 仅展示

# 执行成功

索引类型	适用查询	示例
单列索引	等值/范围过滤	`WHERE category = ?`
复合索引	多条件组合	`WHERE category = ? AND price > ?`
部分索引	条件子集	`WHERE deleted_at IS NULL`
覆盖索引	避免回表	`INCLUDE (name, price)`

(2) 连接池调优

▶ 示例：生产级连接池配置

PYTHON

engine = create_async_engine(
    DATABASE_URL,
    pool_size=25,           # Persistent connections per worker
    max_overflow=10,        # Extra connections when pool exhausted
    pool_timeout=30,        # Wait time for available connection
    pool_recycle=1800,      # Recycle connections after 30 min
    pool_pre_ping=True,     # Test connection before use
    echo=False,             # Disable SQL logging in production
)

输出：

TEXT 📖 仅展示

# 执行成功

参数	推荐值	说明
`pool_size`	CPU cores × 2 + 1	基础连接数
`max_overflow`	pool_size × 0.5	突发连接上限
`pool_timeout`	30s	等待超时
`pool_recycle`	1800s	防止 MySQL/PG 断开空闲连接
`pool_pre_ping`	True	防止使用已断开连接

6. 并发模型优化

(1) Workers 配置

▶ 示例：Gunicorn + Uvicorn Workers

BASH

# Production: Gunicorn manages Uvicorn workers
gunicorn app.main:app \
    --workers 9 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --timeout 120 \
    --graceful-timeout 30 \
    --access-logfile - \
    --error-logfile -

输出：

TEXT 📖 仅展示

# 命令执行成功

DOCKERFILE

# In Dockerfile
CMD ["gunicorn", "app.main:app", \
     "--workers", "9", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

配置	公式/值	说明
Workers 数量	`(2 × CPU) + 1`	4 核 = 9 workers
worker-class	UvicornWorker	ASGI worker
timeout	120s	Worker 超时被杀
graceful-timeout	30s	优雅关闭等待时间
max-requests	10000	自动重启防内存泄漏

❓ 常见问题

Q 如何定位性能瓶颈？

A 用工具链——cProfile 分析 CPU 热点、py-spy 实时采样、Prometheus 监控延迟分布、慢查询日志分析数据库。先测量再优化。

Q run_in_executor 的线程池有多大？

A 默认 min(32, os.cpu_count() + 4)。CPU 密集型任务用 ProcessPoolExecutor 避免受 GIL 限制。

Q orjson 与 Pydantic 兼容吗？

A 兼容。FastAPI 的 default_response_class=ORJSONResponse 自动用 orjson 序列化 Pydantic 模型的 model_dump() 输出。

Q 连接池和 Workers 数量怎么协调？

A 每个 Worker 有自己的连接池。9 Workers × 25 连接 = 225 个数据库连接。PostgreSQL 默认 max_connections=100，需调高或降低 pool_size。

Q GZip 压缩有性能开销吗？

A 压缩消耗 CPU，但减少网络传输时间。对 > 1KB 的 JSON 响应，压缩收益远大于 CPU 开销（压缩比通常 5-10 倍）。

Q 如何验证优化效果？

A 用 locust/k6 进行负载测试，对比优化前后的 QPS、P50/P95/P99 延迟、CPU/内存占用。每次只改一个变量。

📖 小节

性能优化三层模型：应用层（async/序列化）→ 数据库层（索引/连接池）→ 基础设施层（Workers/LB）
同步 I/O 用 run_in_executor 包装避免阻塞事件循环
orjson 替代 json 序列化快 3-10 倍，GZipMiddleware 压缩大响应
数据库索引按查询模式设计（单列、复合、部分索引），连接池大小与 Workers 协调
Gunicorn + Uvicorn Workers：(2 × CPU) + 1 个 Worker，K8s 弹性伸缩到百万 QPS

📝 作业

基础题（难度⭐）：安装 orjson，将 FastAPI 的默认响应类改为 ORJSONResponse，用 curl 对比 json 和 orjson 的响应时间。提示：default_response_class=ORJSONResponse
进阶题（难度⭐⭐）：为 PriceTracker 的 products 表添加索引（category, created_at, category+base_price 复合索引），对比添加索引前后相同查询的执行时间。提示：Index("ix_name", "col1", "col2") + EXPLAIN ANALYZE
挑战题（难度⭐⭐⭐）：完整性能优化——orjson 序列化 + GZipMiddleware + 连接池调优 + Gunicorn 4 Workers，用 locust 跑负载测试对比优化前后 QPS 和 P99 延迟。提示：locust -f locustfile.py --host=http://localhost:8000

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: 性能优化 — 从百到百万 QPS

1. 你将学到

2. Alice 的真实故事

(1) 痛点：单机 500 QPS 不够用

(2) 系统化性能优化的解法

(3) 收益

3. 性能优化层级

(1) 三层优化模型

(2) QPS 提升路径

4. 应用层优化

(1) 异步阻塞排查

▶ 示例：同步代码阻塞事件循环

▶ 示例：run_in_executor 包装同步代码

(2) JSON 序列化优化

▶ 示例：orjson 替代 json

▶ 示例：GZipMiddleware 压缩

5. 数据库层优化

(1) 索引策略

▶ 示例：PriceTracker 索引设计

(2) 连接池调优

▶ 示例：生产级连接池配置

6. 并发模型优化

(1) Workers 配置

▶ 示例：Gunicorn + Uvicorn Workers

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好