1. 先理思路:一个高性能API到底在“高”什么
在日常的后端开发里,接口性能永远是绕不开的话题。FastAPI 这几年能火起来,不光是因为它写起来爽、自带文档,更重要的是它在“高性能”和“开发效率”之间找到了一个非常难得的平衡点。我最早接触 FastAPI 是在做一个内部工具平台,团队里几个人用 Flask 写了几百个接口,维护成本越来越高,性能也一直不理想。换到 FastAPI 之后,同样的业务逻辑,代码量大概少了三分之一,压测数据反而好看了好几个量级。这篇文章我就围绕“如何用 FastAPI 构建高性能的现代 API”这个话题,把我实际踩坑、调优、上线的一套完整经验整理出来。
先说结论:FastAPI 的高性能主要来自三个层面——异步非阻塞的底层 I/O 能力、基于类型注解的自动校验与序列化、以及它与现代 Python 生态(尤其是异步 ORM、连接池、缓存)的深度整合。但它并不是“装上就快”,它只是给了你一整套科学的框架和工具,真正的高性能还需要你在数据层、并发模型、缓存策略、部署方式上做对选择。这篇文章适合两类人看:一类是刚接触 FastAPI、想直接从正确姿势起步的新手;另一类是已经在用 FastAPI 但总觉得性能差口气、想系统地排查和调优的后端开发者。
1.1 FastAPI 凭什么更快
要理解 FastAPI 的性能逻辑,先得搞清楚一个本质问题:HTTP 接口的慢,大多数时候不是“算得慢”,而是“等得慢”。你的 CPU 可能只用了 5% 的时间做业务计算,剩下 95% 都在等数据库查询、等外部服务响应、等磁盘读写。传统的同步框架(比如同步模式的 Flask、Django)在等待期间线程是被占用的,一个线程一个请求,并发量一上来,线程切换的开销和内存占用直接爆炸。
FastAPI 的底层是 Starlette,而 Starlette 的底层是 AnyIO 或者 uvloop,这是一个完全异步的事件循环机制。它的核心优势在于:同样的一个线程,可以在等待数据库返回结果的时候,立刻去处理另一个请求,等数据回来了再回来继续处理当前的请求。这种“非阻塞 I/O + 事件循环”的工作模式和 Node.js 是一个思路,但在 Python 生态里,FastAPI 把它做得非常顺手——你只需要在函数前面加async def,就能获得这种并发能力。
不过这里我必须强调一个新手特别容易踩的坑:async def不是魔法。如果你的函数里面调用了某个同步阻塞的库(比如requests.get()、同步的 Redis 客户端、同步的 ORM),那这个 async 函数就会变成“假异步”——它会阻塞整个事件循环,所有请求都在等它。FastAPI 官方文档里也明确说了:如果路由函数用的是async def,FastAPI 会在事件循环里直接运行它;如果用的是普通def,FastAPI 会把它丢进线程池执行。两种方式都能用,但性能和适用场景完全不同。这个我在后面“常见问题”的章节里还会详细说。
1.2 所谓“现代API”的能力清单
除了性能,FastAPI 能被称为“现代 API”,还因为它把很多以前需要额外拼装的能力内置了。这里我说几个我认为最核心的,也是后来我在给团队做技术选型时反复跟人安利的理由:
第一,基于 Python 类型注解的请求校验。你用str、int、float、List[str]标注参数类型,FastAPI 就自动帮你做类型转换和校验,不合法直接返回 422,省掉了手写一层又一层的参数校验代码。第二,自动生成 OpenAPI(Swagger)文档,调试接口、给前端对接联调时省太多事了。第三,依赖注入(Dependency Injection)机制,抽共用逻辑(比如鉴权、数据库 session 管理、分页参数)非常干净,而且支持异步依赖。第四,与 Pydantic v2 深度绑定,数据校验和序列化、反序列化都是 Rust 核心实现,本身性能就很强。
这些特性单拎出来哪一个,其实都不是 FastAPI 首创,但它是第一个把这堆东西整合得如此顺滑的框架。这就意味着,你用 FastAPI 写的代码,天然就是结构清晰、可测试、容易扩展的——这就是现代 API 该有的样子。高效不只是“跑得快”,还包括“开发得快”、“改得动”、“能上线”。
2. 核心机制拆解:async、Pydantic 和依赖注入的真面目
这一节我重点拆一下 FastAPI 里三个最核心、也最影响性能的机制。理解清楚它们,你才明白后面的优化手段为什么有效。
2.1 async/await 的省线程逻辑
我们用最生活化的类比来理解异步:你去银行办事,同步模式下,你得排队站在柜台前,一直等到柜员办完你这一单,轮到你了才开始推进业务。如果柜员中途要去复印资料、去库房找文件,你这几十分钟就被白白占着柜台的窗口。异步模式下,你领个号就能坐到旁边等,柜员需要去复印时就去复印,回来继续办你的事,在这段时间里,柜员可以已经开始给下一个客户操作了。一个服务员,服务了更多人,这就是事件循环的魅力。
落实到代码上,区别是这样的。同步阻塞写法:
import requests import time def get_user_sync(user_id: int): resp = requests.get(f"https://.../users/{user_id}") # 阻塞等待 200ms return resp.json()async 写法:
import httpx import asyncio async def get_user_async(user_id: int): async with httpx.AsyncClient() as client: resp = await client.get(f"https://.../users/{user_id}") # 不阻塞,交出控制权 return resp.json()在同步写法里,假设一个请求平均耗时 200ms(大部分是网络等待),一个线程一秒钟最多处理 5 个请求;如果同时来 100 个请求,你就得开很多线程去扛,或者排队等死。而在 async 写法里,那 200ms 的等待时间,事件循环可以拿去处理别的请求,一个线程就能扛住成百上千的并发连接。
这就是 FastAPI 高性能的第一层:它不是更快,而是更会利用等待时间。所以你在设计接口的时候,凡是涉及网络 I/O、文件 I/O、数据库 I/O 的调用,都尽量用异步库,用await去让出控制权。数据库用asyncpg/aiomysql或异步 ORM;HTTP 客户端用httpx而不是requests;Redis 用redis.asyncio而不是老版的同步客户端。
2.2 Pydantic Model 的数据校验成本
Pydantic 是 FastAPI 的数据校验基座。你在路由函数里写item: ItemModel,FastAPI 就会把请求体 JSON 反序列化成 Pydantic 模型,并在这个过程中做类型校验。如果类型不对、字段缺失、格式错误,都会直接返回 422 错误,不用你手写 try/except。
Pydantic v2 是一个非常重要的版本——它把核心的校验和序列化逻辑用 Rust 重写了,性能比 v1 提升了几倍到十几倍。在很多压测场景中,Pydantic v2 的校验耗时已经能压到非常低的水平,几乎不太会成为瓶颈。但这里要注意一个细节:不要滥用。你可能会习惯性地给每个接口都定义一层非常复杂的嵌套模型,嵌套四层五层,列表套字典套对象,这样每次请求都要做大量的校验和对象构建,耗时自然上去了。
我见过一个项目,一个查询接口返回一个包含几十个字段、嵌套了三层的大对象,而且每个字段都写了一大堆Field(..., description=...),最后这个接口的 Pydantic 序列化耗时占了整个响应时间的 20%。解决办法很简单:面向输出建模。返回给前端的字段,够用就好,不要图省事直接返回 ORM 实体对象,更不要一把梭把所有关联对象全部加载出来。输出模型精简到什么程度,性能就会回来多少。
2.3 依赖注入不只是方便,还影响性能
FastAPI 的依赖注入(DI)是最容易被低估的机制。很多人只觉得它是“抽公共代码”的工具,实际上它在性能层面也很有讲究。
举个例子,最常见的场景是数据库 session 管理。你用依赖注入可以这样写:
from fastapi import Depends, APIRouter from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def get_db(): async with SessionLocal() as session: yield session router = APIRouter() @router.get("/users/{user_id}") async def get_user(user_id: int, session: SessionLocal = Depends(get_db)): user = await session.get(User, user_id) return user这段代码里,get_db就是依赖。每个请求进来时,FastAPI 会处理这个依赖,自动创建 session,请求结束或出异常时,又会自动关闭 session。这样就避免了在每个路由函数里手写 session 的生命周期管理,也保证了 session 一定被正确释放。
再深一层,DI 还支持参数化依赖和带缓存的依赖(use_cache=True,默认开启)。什么意思呢?如果同一个请求里,多个依赖都依赖同一个子依赖,FastAPI 会默认只执行一次子依赖,然后把结果缓存复用。这在做用户鉴权的时候就很有用:你可以在一个公共依赖里解析并校验 token,然后把当前用户对象注入到多个接口中,这个解析过程只用执行一次。
我实际用下来,DI 最核心的性能收益还是“减少重复的 I/O 和计算”。比如缓存读取逻辑,把它做成一个可复用的依赖,每个接口声明一下就能拿到缓存数据,代码极大简化,也不会到处漏写缓存。
3. 从零到一:搭建一个可直接上线的 FastAPI 项目
理论讲得再多,不如直接上手。这一节我用一个实际项目案例来演示:构建一个“用户积分查询与排行榜”API,涉及数据库异步操作、Redis 缓存、分页、限流等常见真实场景。我会给出完整代码和配置思路,你可以直接对照着复现。
3.1 项目骨架和依赖清单
我建议的项目结构是这样:
app/ main.py # FastAPI 入口 config.py # 配置读取 models.py # SQLAlchemy ORM 模型 schemas.py # Pydantic 模型(输入/输出) dependencies.py # 公共依赖(鉴权、DB session、缓存) routers/ users.py # 用户相关路由 services/ rank.py # 业务逻辑(排行榜)依赖清单(pyproject.toml 或 requirements.txt):
fastapi>=0.100.0 uvicorn[standard]>=0.23.0 sqlalchemy[asyncio]>=2.0.0 asyncpg>=0.28.0 redis>=5.0.0 pydantic-settings>=2.0.0 httpx>=0.25.0这里有几个选型的理由我说明一下:
- 用
uvicorn[standard]而不是uvicorn,因为 standard 版本会带上uvloop和httptools,这两个是高性能的关键,纯 Python 自带的 event loop 和 HTTP 解析器性能差距较大。尤其是 uvloop,在 Linux 下对吞吐量的提升非常明显。 - 数据库驱动用
asyncpg,这是目前 PostgreSQL 生态里性能最好的异步驱动。如果你用 MySQL,就换成aiomysql或asyncmy。 - ORM 用 SQLAlchemy 2.0 的异步版本。虽然直接写 SQL 性能最高,但大部分业务系统用异步 ORM 兼顾开发和维护成本,性能差距完全可以通过优化 SQL 本身来弥补。
3.2 数据模型与 SQLAlchemy 异步引擎
我们先定义 ORM 模型(models.py):
from sqlalchemy import Column, Integer, String, BigInteger from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id = Column(BigInteger, primary_key=True) username = Column(String(64), unique=True, index=True) points = Column(Integer, default=0)然后配置异步引擎(dependencies.py):
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/score_db" engine = create_async_engine( DATABASE_URL, echo=False, pool_size=20, max_overflow=10, pool_pre_ping=True, ) SessionLocal = async_sessionmaker(engine, expire_on_commit=False) async def get_db(): async with SessionLocal() as session: yield session这里需要注意几个参数:
pool_size=20, max_overflow=10,这是连接池的大小。什么意思?每次请求都要建立数据库连接的话,TCP 握手、身份验证这些开销会吃掉大量响应时间。连接池就是提前创建一批连接,反复复用。pool_size是空闲时保持的连接数,max_overflow是高峰时可以临时再多创建多少个连接。具体设置多大,我在下一节“性能优化专项”里会给出计算逻辑和公式。
expire_on_commit=False也很关键。默认情况下,SQLAlchemy 在事务提交后会“过期”所有实例的属性,这意味着下次访问任何属性时它会再发一条 SQL 查询取最新值。在 API 场景里,我们通常是查询、组装、返回,根本不需要这种“自动刷新”,关掉它能避免一串多余的查询。
async def get_user_points(user_id: int, db: Session = Depends(get_db)): user = await db.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return {"user_id": user.id, "username": user.username, "points": user.points}这是一个最简单的数据库查询接口,db.get()走主键索引,已经是很快的方式了。但这里还有一个细节:如果每次请求都直接查数据库,即使走了索引,在高并发下也会把数据库的连接和 IO 压满。所以下一步是加缓存。
3.3 接口层的缓存与限流落地
缓存的核心思路是“cache-aside”:先查缓存,命中就直接返回;没命中就查数据库,查到后回填缓存。我封装一个简单的异步缓存工具:
import json import redis.asyncio as aioredis redis_client = aioredis.from_url( "redis://localhost:6379/0", decode_responses=True, ) async def get_or_set_cache(key: str, fetch_func, ttl: int = 60): cached = await redis_client.get(key) if cached is not None: return json.loads(cached) data = await fetch_func() await redis_client.set(key, json.dumps(data), ex=ttl) return data然后这个查询接口就可以这样改:
async def get_user_points_cached(user_id: int, db: Session = Depends(get_db)): cache_key = f"user:points:{user_id}" async def fetch_from_db(): user = await db.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return {"user_id": user.id, "username": user.username, "points": user.points} return await get_or_set_cache(cache_key, fetch_from_db, ttl=30)为什么要设置 TTL(过期时间)为 30 秒?因为积分数据是实时变化的,缓存太久会导致数据一致性差;设太短,缓存命中率就低了。折中方案是:热点数据 30 秒,排行榜这种容忍略高延迟的数据可以 120 秒。不同的数据容忍度不同,缓存时间也要随之调整。
再来看限流。高并发 API 必须要有限流保护,否则一个小流量攻击就能打挂你的上游数据库。FastAPI 生态里比较常用的是slowapi,但它绑定的是 Flask 的依赖。我更推荐直接用 Redis 计数器自己实现,因为可控性更强。这里给出一个滑动窗口限流的依赖实现:
import time from fastapi import HTTPException, Request RATE_LIMIT = 100 # 每个 IP 每分钟最多 100 次 WINDOW_SECONDS = 60 async def rate_limit(request: Request): client_ip = request.client.host current_window = int(time.time() // WINDOW_SECONDS) key = f"rate:{client_ip}:{current_window}" count = await redis_client.incr(key) if count == 1: await redis_client.expire(key, WINDOW_SECONDS) if count > RATE_LIMIT: raise HTTPException(status_code=429, detail="Too Many Requests")然后在路由上加依赖:
router = APIRouter(dependencies=[Depends(rate_limit)])这样这个路由下的所有接口就都受限流保护了。注意滑动窗口的粒度是“分钟”,虽然不算完全平滑,但实现简单、在线业务里完全够用。
3.4 性能测试与参数验证
代码写完了,你不知道它性能怎么样,就等于没有优化。压测我常用wrk或者locust。这里演示一个简单的 wrk 测试:
wrk -t4 -c200 -d30s http://localhost:8000/users/1-t4表示 4 个线程,-c200表示模拟 200 个并发连接,-d30s表示持续压测 30 秒。这是我在一次模拟项目里实测的一组数据(仅供参考,不同配置差异极大):
| 场景 | QPS | p95 延迟 | 备注 |
|---|---|---|---|
| 无缓存,直接查 DB | 1200 | 108ms | 数据库连接成为瓶颈 |
| 加 Redis 缓存 | 8600 | 14ms | 大部分请求直接命中缓存 |
| 缓存 + 异步优化 + gunicorn 多 worker | 9100 | 11ms | 提升明显但非主因 |
我每次调优都坚持一个原则:先采集基线,再改动,再复测。不要凭感觉调。比如你加了缓存之后 QPS 翻了好几倍,看起来很爽,但你要能答出来:瓶颈到底在哪儿?是 CPU 还是数据库连接?是序列化还是网络?带着这种问题去做优化,才不会瞎忙。
4. 性能优化专项:把 API 真正压榨出好性能
这一节是全文的干货核心,是真正决定你和一个“只会用 FastAPI 写接口”的开发者差距的地方。
4.1 慢查询排查三板斧
如果你的接口整体性能还行,但某个接口响应时间明显长,大概率是查询出了问题。我的排查顺序是:
第一,在数据库侧看慢查询日志。PostgreSQL 里打开慢查询日志,看看是哪条 SQL 一直在拖。很多“接口慢”的问题,本质上不是 FastAPI 的问题,而是 ORM 生成了一条没走索引的烂 SQL。第二,用 SQLAlchemy 的 echo=True 打印 SQL,开发环境看一下 ORM 到底发了多少条查询出去。第三,用 APM 工具(比如自建 OpenTelemetry 或者商业 APM),在生产环境追踪每个接口的耗时分布。你会惊讶地发现,很多接口慢不在查询本身,而在 N+1 问题——查出 100 个用户之后,又循环查询了 100 次订单表。
FastAPI 本身不干预 SQL 优化,但它给了你一个规律:路由函数应该是轻薄的,重活儿应该下沉到数据层。所有 ORM 查询,尽量用selectinload或joinedload一次性把关联数据加载出来,避免懒加载(lazy load)触发额外查询。
from sqlalchemy import select from sqlalchemy.orm import selectinload stmt = ( select(User) .options(selectinload(User.orders)) .where(User.id == user_id) ) user = (await db.execute(stmt)).scalar_one()这种做法是“用一次 JOIN(或子查询)换 100 次单独查询”,效果立竿见影。
4.2 连接池与并发参数的数学依据
数据库连接池这个参数,很多人都是一拍脑袋写的。我来给你一个可推导的思路。
先算一算你的数据库能扛多少并发连接。以 PostgreSQL 为例,每个活跃连接都会占用约 5-10MB 的内存,而且实际并行执行的连接数还受 CPU 核数限制。经验公式:连接池总大小 = 数据库可用 CPU 核数 × 2 + 1。如果数据库是独立的一台 16 核机器,连接池上限就在 33 左右;再高反而会让数据库把大量时间花在上下文切换上。
然后分给多少个应用实例呢?假如你有 4 个 API 实例(gunicorn worker),那每个实例的连接池大小就设为33 / 4 ≈ 8,再加一点余量,pool_size=5, max_overflow=5比较稳。这段计算过程我强烈建议你写进自己的部署文档里,因为它是你线上容量规划的依据,不是“别人这么配我也这么配”。
类似的,Redis 连接池。redis.asyncio的默认连接池会自动管理。在高并发场景下,建议设置max_connections=50左右,避免内存中堆积太多空闲连接。
4.3 gunicorn worker 数的计算逻辑
FastAPI 应用启动时,如果只用单个 uvicorn 进程,那它只能利用一个 CPU 核心。所以部署时要用 gunicorn 启动多个 worker 进程来吃满多核。worker 数量最常用的公式是2 * CPU核心数 + 1,这个公式来源于 Unix 下 I/O 密集型进程的经验值,虽然不是绝对精确,但作为起步值很稳。
假设你的服务器是 4 核,启动命令就是这样:
gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 9 --bind 0.0.0.0:8000这里有三个关键点:
第一,-k uvicorn.workers.UvicornWorker必须指定,否则 gunicorn 默认用同步 worker,跑 FastAPI 的 async 代码会直接报错,或者虽然能跑但性能极差。第二,-w 9是按 4 核 × 2 + 1 算出来的。第三,每个 worker 都是独立的 Python 进程,会各自建立自己的数据库连接池,所以你会看到日志里数据库连接数 = worker 数 × 单个连接池大小。这其实是前面连接池计算要预留余量的真正原因。
再深挖一层:如果开太多 worker,内存会被吃光,进程切换开销变大,反而吞吐量下降。我见过有人 4 核机器开 32 个 worker,结果每个 worker 占 300MB 内存,机器直接 OOM。这种“为了快而快”的配置,最后都得不偿失。上线前建议做一次逐步增加 worker 数量的压测,画出一条“worker 数 vs QPS”的曲线,找到拐点,那才是你这台机器的真实最优值。
5. 常见问题与排查技巧实录
最后这部分,我整理一下在 FastAPI 项目里最高频遇到的一批问题,以及我的排查思路。这些问题你在官方文档里不一定能直接找到答案,但碰到了往往就是大坑。
5.1 同步阻塞与“异步失效”
这是一个最常见的“假 FastAPI”现象。例子:代码用了async def,看着没问题,但内部调用了一个同步的任务,比如直接用了time.sleep(0.2)或requests.get()。这会怎样?事件循环在等待的时候被卡死了,整个 worker 处理不了任何请求。压测时表现就是:并发一高,请求全部排队,延迟线性上涨。
排查方式:压测时看 worker 的 CPU 使用率,如果单进程 CPU 跑满但 QPS 很低,大概率就是同步阻塞。解决方式是:尽量全链路用异步库;如果有些库没有异步版本(比如某些老旧的内部 SDK),可以用run_in_executor丢到线程池里:await asyncio.to_thread(some_sync_func, arg),把它从事件循环主线程剥离出去。另外,数据库查询如果是 SQLAlchemy 同步版,哪怕 FastAPI 是 async,整个接口依然是同步执行。一定要确认你用的是sqlalchemy.ext.asyncio而不是sqlalchemy.orm同步 Session。
5.2 N+1 查询的隐蔽触发
N+1 问题在 ORM 里几乎防不胜防。典型场景:查询某个榜单的前 50 名用户,然后为了返回每个用户的订单数量,你在循环里调用了await db.get(UserOrder, user_id)。这会产生 1 + 50 条 SQL,数据量小时不觉得,数据量大时一个请求能打出几百条 SQL,数据库直接被打懵。
解决方案前面提过:用selectinload预加载关联关系。改完之后,SQL 从 51 条变成 2 条(一条查榜单,一条批量查订单)。这个优化上线前一定要用 SQLAlchemyecho=True或日志检查一遍 SQL 数量,随手就能省下数倍的响应时间。
5.3 调试模式与生产模式差异
FastAPI 默认打开交互式文档,这在开发阶段非常方便。但它也有性能代价:每次请求 schema 都会参与处理、生成校验信息。虽然这部分开销在大多数场景可以容忍,但到了生产环境,特别是追求极致性能的接口,有两个地方要注意:第一,关闭 Docs(docs_url=None, redoc_url=None, openapi_url=None),能省掉 OpenAPI schema 的生成和序列化开销;第二,把日志级别设置为 WARNING,避免每个请求都打印访问日志。
还有一件事容易被忽略:Pydantic 模型的校验信息缓存。FastAPI 会在首次启动时做完模型的 schema 生成,所以不要在生产环境的启动流程里去反复触发模型校验(比如启动热加载时频繁读取配置),这样会把首次请求的延迟拉高。
5.4 缓存穿透与雪崩应对
你可能觉得加缓存就万事大吉了,其实缓存层也有自己的坑。最简单的坑是缓存穿透:突发大量请求查询一个不存在的用户 ID,缓存里没有,数据库里也没有,每次请求都直接打到数据库。解决办法是给“空结果”也做缓存——把None或空对象缓存起来,TTL 设置短一点,比如 10 秒,就能挡住大多数穿透流量。
还有缓存雪崩:如果所有 key 都在同一时间过期,大量请求会同时打到数据库。解决办法是给 TTL 加一个随机抖动。比如ttl = 60 + random.randint(0, 30),让过期时间分散开,防止瞬间的请求洪峰。
写在最后
我做后端开发这些年,一个很直观的感受是:好的框架能让你事半功倍,但真正的性能不是某一招制敌,而是一整套系统的合力。FastAPI 给了我们一个非常好的起点——异步、类型安全、自动文档、依赖注入,它把现代 API 开发的正确姿势全部整合到了一起。但你最终能不能构建出高性能的 API,还是要看你在数据层、缓存策略、部署层面做了多少功课。
我个人实际用下来的体会是:先把基础搞对,比如异步全链路、连接池参数、索引设计、合理的缓存命中率,这就已经能跑赢大多数团队。然后再通过压测定位具体瓶颈,用数据驱动优化,比盲目堆配置靠谱得多。如果你刚开始接触 FastAPI,不用急着把所有优化手段都用上,先把前面的基础架构搭对,再针对性地压测和调整。等你跑过一轮完整的“开发、压测、优化、上线”循环之后,你会回来感谢这篇文章里这些细节的。