FastAPI 性能：你可能忽视的隐藏线程池开销

Source: Dev.to

理解问题

FastAPI 是一个用于在 Python 中构建高性能 API 的绝佳框架。它的异步能力、自动验证以及出色的文档让开发体验非常愉快。然而，一个细微的性能问题常常被忽视：对同步依赖的不必要的线程池委派。

FastAPI 如何处理依赖

FastAPI 区分 async 与 sync 可调用对象：

• async def 函数 – 直接在事件循环中执行。
• def 函数 – 通过 anyio.to_thread.run_sync 发送到线程池。

这种行为同样适用于路径操作函数和依赖。FastAPI 在内部执行一个简化的检查：

import asyncio
from anyio import to_thread

# 简化的 FastAPI 逻辑
if asyncio.iscoroutinefunction(dependency):
    # 直接在事件循环中运行
    result = await dependency()
else:
    # 发送到线程池
    result = await to_thread.run_sync(dependency)

由于类构造函数（__init__）始终是同步的，基于类的依赖总是会被路由到线程池。

线程池开销

• 默认线程池大小： 40 个线程。
• 每一次线程池执行都会产生上下文切换、线程同步，以及在所有线程忙碌时可能的排队。

示例：多个基于类的依赖

from fastapi import Depends, FastAPI

app = FastAPI()

class QueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
    return {"q": params.q, "skip": params.skip, "limit": params.limit}

每一次请求都会在线程池中创建一个 QueryParams 实例，尽管它只做了简单的赋值。

如果一个端点拥有多个此类依赖，开销会成倍增加：

@app.get("/complex-endpoint/")
async def complex_operation(
    auth: AuthParams = Depends(),
    query: QueryParams = Depends(),
    pagination: PaginationParams = Depends(),
    filters: FilterParams = Depends(),
):
    pass  # 四个依赖 → 四个线程池任务

在 100 个并发请求的情况下，会产生 400 个线程池任务被排队，但只有 40 个能够同时运行，从而导致延迟峰值。

真实世界的影响

• 拥有 50 条端点的 API
• 每条端点平均 3 个基于类的依赖
• 每秒 1 000 次请求

→ 每秒约 150 000 次不必要的线程池操作。即使每次操作很快，累计的开销也可能成为瓶颈。

解决方案：fastapi-async-safe-dependencies

一个轻量级库，用于将某些依赖标记为可以安全地直接在事件循环中运行，绕过线程池。

安装

pip install fastapi-async-safe-dependencies

基本用法

from fastapi import Depends, FastAPI
from fastapi_async_safe import async_safe, init_app

app = FastAPI()
init_app(app)  # 初始化库

@async_safe  # 标记为在 async 环境下安全
class QueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
    return {"q": params.q, "skip": params.skip, "limit": params.limit}

有什么变化？

1. 启动时调用 init_app(app)。
2. 使用 @async_safe 装饰依赖类。

工作原理

当类被 @async_safe 装饰时，库会生成一个 async 包装器：

# @async_safe 生成的简化包装器
async def _wrapper(**kwargs):
    return YourClass(**kwargs)  # 直接调用构造函数

因为包装器是协程，asyncio.iscoroutinefunction 会返回 True，于是 FastAPI 会直接在事件循环中运行它——不再涉及线程池。

init_app() 会遍历所有路由和依赖，将类引用替换为这些包装器。包装器本身不执行 await，只是在同步构造函数上快速运行，这在构造函数不阻塞时是安全的。

支持继承

from fastapi_async_safe import async_safe

@async_safe
class BaseParams:
    def __init__(self, limit: int = 100):
        self.limit = min(limit, 1000)

class QueryParams(BaseParams):
    def __init__(self, q: str | None = None, **kwargs):
        super().__init__(**kwargs)
        self.q = q

如果子类 需要 线程池执行（例如进行 I/O），使用 @async_unsafe 标记：

from fastapi_async_safe import async_unsafe

@async_safe
class BaseParams:
    pass

@async_unsafe  # 将被发送到线程池
class HeavyParams(BaseParams):
    def __init__(self):
        self.data = some_blocking_operation()

全局 Opt‑In

init_app(app, all_classes_safe=True)  # 将所有基于类的依赖视为 async‑safe
# 仅在例外情况下使用 @async_unsafe

与同步函数一起使用

装饰器同样适用于普通函数：

from fastapi_async_safe import async_safe

@async_safe
def get_common_params(q: str | None = None, skip: int = 0, limit: int = 100) -> dict:
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(params: dict = Depends(get_common_params)):
    return params

基准测试与结果

最佳实践

何时使用 @async_safe

✅ 简单数据类
✅ 参数校验类
✅ 配置对象
✅ 非阻塞的工具函数
✅ Pydantic 模型包装器

❌ 不要 用于：

• 数据库查询
• 文件 I/O
• 外部 API 调用
• CPU 密集型计算
• 任何真正阻塞事件循环的操作

采纳策略

1. 从小处开始 – 先在调用最频繁的端点上使用。
2. 监控 – 确认延迟得到改善且没有回归。
3. 逐步扩展 – 逐渐标记更多依赖为 async‑safe。
4. 考虑全局 Opt‑In – 当足够自信时，使用 all_classes_safe=True。

测试

现有测试保持不变：

import pytest
from fastapi.testclient import TestClient

def test_endpoint():
    client = TestClient(app)
    response = client.get("/items/?q=test&limit=50")
    assert response.status_code == 200
    assert response.json()["q"] == "test"

注意事项

• 过早优化 – 只有在观察到性能问题时才采用。
• 阻塞依赖 – 保持在线程池中（使用 @async_unsafe）。
• 先进行分析 – 使用 uvicorn --log-level debug 或外部分析工具确认瓶颈后再使用本库。