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 defFastAPI 会在事件循环里直接运行它如果用的是普通defFastAPI 会把它丢进线程池执行。两种方式都能用但性能和适用场景完全不同。这个我在后面“常见问题”的章节里还会详细说。1.2 所谓“现代API”的能力清单除了性能FastAPI 能被称为“现代 API”还因为它把很多以前需要额外拼装的能力内置了。这里我说几个我认为最核心的也是后来我在给团队做技术选型时反复跟人安利的理由第一基于 Python 类型注解的请求校验。你用str、int、float、List[str]标注参数类型FastAPI 就自动帮你做类型转换和校验不合法直接返回 422省掉了手写一层又一层的参数校验代码。第二自动生成 OpenAPISwagger文档调试接口、给前端对接联调时省太多事了。第三依赖注入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(fhttps://.../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(fhttps://.../users/{user_id}) # 不阻塞交出控制权 return resp.json()在同步写法里假设一个请求平均耗时 200ms大部分是网络等待一个线程一秒钟最多处理 5 个请求如果同时来 100 个请求你就得开很多线程去扛或者排队等死。而在 async 写法里那 200ms 的等待时间事件循环可以拿去处理别的请求一个线程就能扛住成百上千的并发连接。这就是 FastAPI 高性能的第一层它不是更快而是更会利用等待时间。所以你在设计接口的时候凡是涉及网络 I/O、文件 I/O、数据库 I/O 的调用都尽量用异步库用await去让出控制权。数据库用asyncpg/aiomysql或异步 ORMHTTP 客户端用httpx而不是requestsRedis 用redis.asyncio而不是老版的同步客户端。2.2 Pydantic Model 的数据校验成本Pydantic 是 FastAPI 的数据校验基座。你在路由函数里写item: ItemModelFastAPI 就会把请求体 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(postgresqlasyncpg://user:passlocalhost/db) SessionLocal async_sessionmaker(engine, expire_on_commitFalse) 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_cacheTrue默认开启。什么意思呢如果同一个请求里多个依赖都依赖同一个子依赖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.txtfastapi0.100.0 uvicorn[standard]0.23.0 sqlalchemy[asyncio]2.0.0 asyncpg0.28.0 redis5.0.0 pydantic-settings2.0.0 httpx0.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.pyfrom sqlalchemy import Column, Integer, String, BigInteger from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class User(Base): __tablename__ users id Column(BigInteger, primary_keyTrue) username Column(String(64), uniqueTrue, indexTrue) points Column(Integer, default0)然后配置异步引擎dependencies.pyfrom sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker DATABASE_URL postgresqlasyncpg://user:passlocalhost:5432/score_db engine create_async_engine( DATABASE_URL, echoFalse, pool_size20, max_overflow10, pool_pre_pingTrue, ) SessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_db(): async with SessionLocal() as session: yield session这里需要注意几个参数pool_size20, max_overflow10这是连接池的大小。什么意思每次请求都要建立数据库连接的话TCP 握手、身份验证这些开销会吃掉大量响应时间。连接池就是提前创建一批连接反复复用。pool_size是空闲时保持的连接数max_overflow是高峰时可以临时再多创建多少个连接。具体设置多大我在下一节“性能优化专项”里会给出计算逻辑和公式。expire_on_commitFalse也很关键。默认情况下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_code404, detailUser 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_responsesTrue, ) 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), exttl) return data然后这个查询接口就可以这样改async def get_user_points_cached(user_id: int, db: Session Depends(get_db)): cache_key fuser:points:{user_id} async def fetch_from_db(): user await db.get(User, user_id) if not user: raise HTTPException(status_code404, detailUser not found) return {user_id: user.id, username: user.username, points: user.points} return await get_or_set_cache(cache_key, fetch_from_db, ttl30)为什么要设置 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 frate:{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_code429, detailToo 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 秒。这是我在一次模拟项目里实测的一组数据仅供参考不同配置差异极大场景QPSp95 延迟备注无缓存直接查 DB1200108ms数据库连接成为瓶颈加 Redis 缓存860014ms大部分请求直接命中缓存缓存 异步优化 gunicorn 多 worker910011ms提升明显但非主因我每次调优都坚持一个原则先采集基线再改动再复测。不要凭感觉调。比如你加了缓存之后 QPS 翻了好几倍看起来很爽但你要能答出来瓶颈到底在哪儿是 CPU 还是数据库连接是序列化还是网络带着这种问题去做优化才不会瞎忙。4. 性能优化专项把 API 真正压榨出好性能这一节是全文的干货核心是真正决定你和一个“只会用 FastAPI 写接口”的开发者差距的地方。4.1 慢查询排查三板斧如果你的接口整体性能还行但某个接口响应时间明显长大概率是查询出了问题。我的排查顺序是第一在数据库侧看慢查询日志。PostgreSQL 里打开慢查询日志看看是哪条 SQL 一直在拖。很多“接口慢”的问题本质上不是 FastAPI 的问题而是 ORM 生成了一条没走索引的烂 SQL。第二用 SQLAlchemy 的 echoTrue 打印 SQL开发环境看一下 ORM 到底发了多少条查询出去。第三用 APM 工具比如自建 OpenTelemetry 或者商业 APM在生产环境追踪每个接口的耗时分布。你会惊讶地发现很多接口慢不在查询本身而在 N1 问题——查出 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_size5, max_overflow5比较稳。这段计算过程我强烈建议你写进自己的部署文档里因为它是你线上容量规划的依据不是“别人这么配我也这么配”。类似的Redis 连接池。redis.asyncio的默认连接池会自动管理。在高并发场景下建议设置max_connections50左右避免内存中堆积太多空闲连接。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 N1 查询的隐蔽触发N1 问题在 ORM 里几乎防不胜防。典型场景查询某个榜单的前 50 名用户然后为了返回每个用户的订单数量你在循环里调用了await db.get(UserOrder, user_id)。这会产生 1 50 条 SQL数据量小时不觉得数据量大时一个请求能打出几百条 SQL数据库直接被打懵。解决方案前面提过用selectinload预加载关联关系。改完之后SQL 从 51 条变成 2 条一条查榜单一条批量查订单。这个优化上线前一定要用 SQLAlchemyechoTrue或日志检查一遍 SQL 数量随手就能省下数倍的响应时间。5.3 调试模式与生产模式差异FastAPI 默认打开交互式文档这在开发阶段非常方便。但它也有性能代价每次请求 schema 都会参与处理、生成校验信息。虽然这部分开销在大多数场景可以容忍但到了生产环境特别是追求极致性能的接口有两个地方要注意第一关闭 Docsdocs_urlNone, redoc_urlNone, openapi_urlNone能省掉 OpenAPI schema 的生成和序列化开销第二把日志级别设置为 WARNING避免每个请求都打印访问日志。还有一件事容易被忽略Pydantic 模型的校验信息缓存。FastAPI 会在首次启动时做完模型的 schema 生成所以不要在生产环境的启动流程里去反复触发模型校验比如启动热加载时频繁读取配置这样会把首次请求的延迟拉高。5.4 缓存穿透与雪崩应对你可能觉得加缓存就万事大吉了其实缓存层也有自己的坑。最简单的坑是缓存穿透突发大量请求查询一个不存在的用户 ID缓存里没有数据库里也没有每次请求都直接打到数据库。解决办法是给“空结果”也做缓存——把None或空对象缓存起来TTL 设置短一点比如 10 秒就能挡住大多数穿透流量。还有缓存雪崩如果所有 key 都在同一时间过期大量请求会同时打到数据库。解决办法是给 TTL 加一个随机抖动。比如ttl 60 random.randint(0, 30)让过期时间分散开防止瞬间的请求洪峰。写在最后我做后端开发这些年一个很直观的感受是好的框架能让你事半功倍但真正的性能不是某一招制敌而是一整套系统的合力。FastAPI 给了我们一个非常好的起点——异步、类型安全、自动文档、依赖注入它把现代 API 开发的正确姿势全部整合到了一起。但你最终能不能构建出高性能的 API还是要看你在数据层、缓存策略、部署层面做了多少功课。我个人实际用下来的体会是先把基础搞对比如异步全链路、连接池参数、索引设计、合理的缓存命中率这就已经能跑赢大多数团队。然后再通过压测定位具体瓶颈用数据驱动优化比盲目堆配置靠谱得多。如果你刚开始接触 FastAPI不用急着把所有优化手段都用上先把前面的基础架构搭对再针对性地压测和调整。等你跑过一轮完整的“开发、压测、优化、上线”循环之后你会回来感谢这篇文章里这些细节的。