前阵子我接手了一个内部数据处理服务,约三十个接口用Flask写成,参数校验全是手撸代码,接口文档靠Word和Postman维护,调用方一多就开始超时。当时有个同事提议用FastAPI重写一个小模块做验证,结果验证完我们直接把整个服务都迁到了FastAPI上。这篇就是这段时间的完整复盘,包括选型思路、项目骨架、数据库接入、生产部署,以及实际踩过的几个坑。
先给结论:FastAPI真正打动我的不是"给你一个跑得快的新框架",而是它把类型声明、参数校验、接口文档、序列化这一整套事情全部串了起来,代码量明显减少,每条链路都有章可循。至于"高性能"三个字,正确的打开方式是搞清楚它快在哪里、慢在哪里——如果只是把一个同步密集计算型接口原封不动扔进去,性能并不会自动起飞。它能做到比传统同步框架更高的吞吐,靠的是异步I/O、高效的序列化实现和正确的部署方式三者的组合。
这篇文章适合三类人:一是正在做框架选型、想从Flask或Django迁移的团队,二是已经用FastAPI写了接口但还没上生产、想了解部署和调优关键点的开发者,三是对"FastAPI高性能"这个说法半信半疑、想搞明白底层原因的人。接下来讲的都是我在实际项目里验证过的做法,不是从文档里照搬的。
1. 换框架之前,我先搞清了FastAPI的"高性能"到底指什么
1.1 老项目里的三个真实痛点
先说我在老项目里最难受的三件事。
第一,参数校验的代码会无限膨胀。一个创建订单的接口,请求体里可能有十几个字段,每个都要判断类型、判断范围、判断是否可选。Flask里我一般写一个check_order_params()函数,然后每个接口里调用,再手动返回400。字段一多,这个函数比业务逻辑还长。
第二,接口文档永远滞后。当时我们有一份接口文档,但是代码改了字段,文档没跟上,调用方照着文档传参数就会报错。后来只能用Postman维护一份,再写一份部署文档,维护成本非常高。
第三,多来源参数的读取容易出bug。路径参数、查询参数、请求头、JSON请求体,Flask里都要自己取,而且获取方式和错误处理在20个接口里可能就有5种写法,出问题的时候排查特别费劲。
当时评估过几个方向:继续用Flask加各种插件、切Django REST Framework、切FastAPI。最终选FastAPI,本质上是希望用一套统一的声明式写法同时解决以上三个问题。
1.2 FastAPI的"快"主要来自三个层面
很多人一说到FastAPI就觉得"快",但快不是凭空来的。我拆开讲三个来源。
第一层是异步。FastAPI构建在Starlette之上,底层走ASGI协议。和传统的WSGI同步模型不同,当你的接口需要等数据库返回、等下游HTTP响应、等Redis结果时,事件循环可以在这段时间里处理其他请求。这里有个大前提:你的业务必须真的让出了控制权,也就是用了async def加可等待的调用,否则异步只是名字上有,实际还是排队。
第二层是序列化。FastAPI用Pydantic做数据校验和序列化,而且Pydantic v2的核心是用Rust写的,解析速度相比纯Python实现快了好几倍。我实测过同一个复杂嵌套结构,Pydantic v2的校验耗时大约是v1的三分之一左右。这意味着同样的数据量,FastAPI在请求进站和响应出站时的CPU开销更低。
第三层是部署模型。FastAPI官方推荐用Uvicorn配合Gunicorn运行,每个worker进程内部跑一个事件循环,多个worker利用多核CPU并行处理。这个模型比传统的同步worker能撑住更高的并发连接数,尤其在长连接和IO密集型场景下优势明显。
1.3 同步与异步:你得先接受这个框架的边界
这里必须泼一盆冷水:异步不是万能药。如果你的接口内部全是对数据库的同步阻塞调用,即使写了async def,阻塞发生的那一刻,整个事件循环也会被卡住。我见过不少初学者把全部接口都写成async def,结果性能反而比同步还差。
所以我的判断标准很简单:
- 接口内部主要是查询转发、读写数据库、调用其他HTTP服务——适合异步,吞吐会上去。
- 接口内部有大量CPU计算,比如图像处理、复杂算法、文件压缩——异步帮助不大,应该用同步接口甚至单独开进程处理。
FastAPI允许你混着写:async def用于异步场景,普通def用于同步计算。框架会在线程池中执行同步函数,不会阻塞事件循环。我最初迁移时为了省事,把一些CPU密集型接口也写成了async def,压测表现非常差,改成普通def后反而稳定了。这个细节值得每个准备入坑的人记住。
2. 一套可以直接复用的FastAPI项目骨架
2.1 最小依赖和项目结构划分
搭建项目时,我建议直接使用Python 3.10以上版本。原因很现实:新版Python对类型标注的语法支持更完整,FastAPI开发时写的那些|联合类型、list[str]等写法在旧版本里会有兼容问题,没必要在这个地方给自己添堵。
核心依赖只需要两个:fastapi和uvicorn[standard]。不要一上来就安装一堆插件,FastAPI的生态特点是很多功能可以用标准库或一层薄封装解决,装太多反而增加认知负担。
我的项目结构一般长这样:
app/ main.py # 创建FastAPI实例,注册路由 api/ routes/ order.py # 订单相关路由 user.py # 用户相关路由 deps.py # 依赖注入定义 models/ order.py # SQLAlchemy ORM模型 user.py schemas/ order.py # Pydantic请求/响应模型 user.py core/ config.py # 应用配置 database.py # 数据库引擎与会话 services/ # 业务逻辑层这样划分的核心逻辑是:路由只负责接收HTTP请求和返回HTTP响应,业务逻辑放在services里,数据模型和对外Schema严格分离。实际开发中,最容易出现的混乱就是把ORM对象直接当响应返回,导致无关字段泄露给调用方,这个在后面的序列化部分会详细说。
2.2 路由与Pydantic模型:声明式接口的第一竞争力
FastAPI最让我上头的特性,是把接口的参数定义直接写在函数签名里。
看一个简单的商品创建接口对比。Flask时代我的代码大致长这样:手动取JSON、判断字段存在、转类型、校验范围、拼接错误信息。而FastAPI里可以这样写:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app = FastAPI() class ItemCreate(BaseModel): name: str = Field(..., min_length=1, max_length=64) price: float = Field(..., gt=0, le=1000000) tags: list[str] = Field(default_factory=list) @app.post("/items") async def create_item(item: ItemCreate): # 到这里时,item已经完全通过校验 return {"name": item.name, "price": item.price, "tags": item.tags}这个接口一旦定义完,三件事同时完成:请求体自动解析成ItemCreate实例,非法数据自动返回422并附带错误详情,/docs页面自动出现可调试的接口文档。我算过一笔账,同样的功能用Flask实现,光参数校验的代码量就是这边的三到四倍,而且每多一个接口,这部分代码就多一份重复。
路径参数和查询参数同理,声明在函数签名里即可:
@app.get("/items/{item_id}") async def get_item( item_id: int, detail: bool = False, category: str | None = None, ): # item_id自动转成int且校验类型 # category可选,未传时是None ...类型标注在这里不是装饰品,而是数据契约。调用方看/docs里的OpenAPI文档,就能精确知道每个字段的类型和约束,这个问题我在维护Word文档时想了五年都没解决。
2.3 依赖注入:比装饰器更好用的横切逻辑处理方式
很多刚从Flask转过来的人,容易忽略Depends的使用。但它恰恰是FastAPI设计里最有价值的部分之一。
依赖注入解决的是"多个接口都需要同一份前置处理"的问题。比如获取当前用户、创建数据库会话、校验权限。如果用装饰器,逻辑是隐式的——你看函数定义时根本不知道装饰器做了什么;而Depends把依赖写在函数签名里,谁依赖什么都一目了然。
实际项目里最典型的例子是获取当前用户:
from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security = HTTPBearer() def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), ): token = credentials.credentials user = verify_token(token) if user is None: raise HTTPException(status_code=401, detail="invalid token") return user @app.get("/users/me") async def read_me(user: dict = Depends(get_current_user)): return {"user_id": user["id"], "name": user["name"]}这里的好处是,get_current_user可以在任意接口里复用,也可以在写测试时直接替换。你可以把它替换成一个返回固定测试用户的函数,完全不用改接口代码。
依赖注入还能组合。一个接口可以从多个依赖里取数据,每个依赖都可以再依赖其他依赖。这种组合性让复杂业务的前置校验变得非常干净。我自己的经验是:权限校验、会话管理、公共参数读取,这三类逻辑优先用依赖注入,不要写装饰器。
3. 数据库接入:异步会话的生命周期管理是成败关键
3.1 异步引擎与会话工厂的初始化细节
FastAPI没有绑定ORM,我用的是SQLAlchemy 2.0配AsyncPG驱动,这也是目前最常见的组合。初始化时有个容易踩的坑:异步引擎与会话工厂的配置不要散落在各个文件里,最好集中在数据库模块中。
我通常这样配置:
from sqlalchemy.ext.asyncio import ( create_async_engine, async_sessionmaker, ) from sqlalchemy.orm import DeclarativeBase # PostgreSQL + asyncpg 驱动 DATABASE_URL = "postgresql+asyncpg://user:password@host:5432/dbname" engine = create_async_engine( DATABASE_URL, echo=False, pool_size=20, max_overflow=10, pool_pre_ping=True, ) async_session = async_sessionmaker( engine, expire_on_commit=False ) class Base(DeclarativeBase): pass几个参数值得解释一下。pool_size是连接池的常驻连接数,max_overflow是超过池大小时最多还能临时创建的连接数。这个数值要结合Uvicorn worker数量和数据库服务端限制来定,不是越大越好。pool_pre_ping=True会在取连接时先做一次探测,可以避免使用MySQL或PostgreSQL断开后的幽灵连接。expire_on_commit=False很关键,否则事务提交后,你再访问ORM对象的属性时,它会重新发起查询,这在异步环境里容易引发额外性能消耗。
3.2 用依赖注入管理会话生命周期
数据库会话的获取和关闭,是我见过最容易出问题的地方。新手常犯的错误是每个函数里手动await session.close(),一旦某个分支忘了关闭,连接池就会被慢慢耗干。
推荐的做法是把会话的生命周期交给依赖注入管理,利用它的退出清理机制:
from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession async def get_session(): async with async_session() as session: yield session @app.get("/orders") async def list_orders( db: AsyncSession = Depends(get_session), ): result = await db.execute(select(Order)) return result.scalars().all()依赖里的async with会在请求结束时自动关闭会话并释放连接。即使接口内部抛出了异常,会话也能正常退出。这个模式我已经用了很久,生产环境里没有再出现过连接池耗尽的问题。
事务控制方面,我的习惯是明确提交。SQLAlchemy 2.0异步会话默认会开启事务,但提交时机你要自己想清楚。简单查询可以依赖上下文自动管理,但涉及写操作的业务,我会显式调用await db.commit()。更复杂的情况,比如多表写入要求原子性,可以在业务逻辑层使用async with db.begin():手动控制事务边界。
3.3 分页、批量写入与连接参数调优
分页我直接用一个公共依赖来统一处理,避免每个接口各自写一套limit和offset的解析:
from pydantic import BaseModel, Field class PaginationParams(BaseModel): page: int = Field(1, ge=1) page_size: int = Field(20, ge=1, le=100) async def get_pagination( page: int = 1, page_size: int = 20, ): offset = (page - 1) * page_size return offset, page_size这样所有列表接口的分页行为都是一致的。批量写入方面,和单条逐次插入比起来,executemany风格的操作能有非常明显的提升。如果你用的是PostgreSQL,insert语句可以走psycopg驱动提供的批量能力,SQLAlchemy里可以用sqlalchemy.dialects.postgresql.insert配合on_conflict_do_update做upsert,一套写法同时兼顾插入和更新。
连接参数的调优要结合压测数据。我最初按网上推荐的默认值配,压测时发现数据库连接数频繁报警,后来参考了数据库侧pg_stat_activity的观测结果,把pool_size调低了,进程内总连接数才稳定下来。这个事没有统一答案,务必把连接数监控加进你的运营面板。
4. 上线前的性能准备:六个让接口真正变快的实操项
4.1 生产部署:worker数量别拍脑袋
FastAPI的官方地位是"ASGI框架",它自己不处理生产环境的进程管理,这是Uvicorn的活,而Gunicorn则负责管理多个worker进程。我的生产部署配置长这样:
gunicorn app.main:app \ --worker-class uvicorn.workers.UvicornWorker \ --bind 0.0.0.0:8000 \ --workers 4 \ --timeout 60 \ --graceful-timeout 30--worker-class必须是uvicorn.workers.UvicornWorker,如果直接跑同步worker,性能会一落千丈。worker数量上,我的经验是看CPU核数和业务类型。如果是IO密集型,可以设为CPU核数的2到4倍;如果接口里有大量耗时计算,建议worker数不超过物理核数,避免过多上下文切换和内存压力。
部署的时候还有个容易踩的点:--timeout设太短。接口里如果有长时间挂起的异步操作,worker会被Gunicorn当成卡死进程直接杀掉,重试造成更严重的排队。我见过一个报表接口稳定运行但每天固定时间点超时,最后发现就是--timeout设成30秒,而那个接口正常耗时就是45秒。
4.2 序列化开销:response_model不只是生成文档用的
很多人把response_model当成文档装饰品,实际上它对接口性能有直接帮助。声明响应模型后,FastAPI会做两件事:过滤掉未声明字段,同时把结果序列化成声明的结构。
看这个例子:
from pydantic import BaseModel class UserResponse(BaseModel): id: int name: str email: str @app.get("/users/{user_id}", response_model=UserResponse) async def get_user(user_id: int, db: AsyncSession = Depends(get_session)): user = await db.get(User, user_id) return user即使user对象里有密码字段,响应模型也会自动滤掉。这样既安全,也减少了响应体量。
但要注意一个开销问题:如果你查询了10个字段,而响应模型只需要5个,数据库传输阶段仍然会把10个字段数据捞回来。所以性能优化要从两层做:外层用response_model控制输出,内层SQL查询时用select(User.name, User.email...)只取必要列。这两步配合,IO开销和数据传输量才能真正降下来。
4.3 缓存分三层:进程内、API层与数据库查询层
我在压测中发现,系统的性能瓶颈几乎总在数据库和下游服务调用上,框架本身极少成为瓶颈。所以缓存是绕不开的优化手段。
我常用的缓存策略分三层。第一层是进程内缓存,适合那些不常变化又频繁访问的配置数据。Python标准库的functools.lru_cache可以用在同步函数上,用在异步函数上要注意:它默认不会缓存await的结果,你需要用支持异步版本的实现,或者把查询逻辑包在一个同步函数里再用lru_cache。
第二层是API响应缓存,适合接口结果可以短时间复用的情况。我会给这类接口加一个"缓存版本号"参数,有写入操作时递增版本号,让旧缓存自然失效。
第三层是数据库查询缓存,适合高频读但数据变化慢的场景。实现上,先用Redis查缓存键,命中就直接返回,不命中再查询数据库并回填。回填时我会设置合理的过期时间——通常是60秒到300秒之间,具体数值取决于业务对数据新鲜度的容忍度。
4.4 压测五步法与瓶颈定位思路
上线前压测不要只测"能不能跑",流程要足够完整。我的方法通常是五步。
第一步,准备测试数据和调用方模拟——最好用真实接口参数分布,不要全部请求命中最简单的路径。第二步,用压测工具逐步增加并发数,从10并发开始,每次翻倍,观测P95、P99延迟和错误率。第三步,同步观测服务端指标:CPU、内存、文件描述符、数据库连接数、慢查询日志。第四步,瓶颈定位——如果CPU满而数据库没压力,基本是业务代码或序列化在消耗CPU;如果数据库连接排满,就要调连接池或加索引;如果P99延迟高但CPU和DB都正常,问题大概率在下游HTTP调用,重点排查超时设置和连接复用。第五步,针对问题做优化,再重复压测验证。
压测工具我用过wrk、locust和k6。wrk适合单机快速测试;locust适合分布式压测;k6更适合集成到CI里做回归测试。我的建议是至少选一个,把压测脚本纳入仓库,接口改造后跑一遍对比数据。
5. 生产环境里踩过的坑和修复记录
5.1 一个同步HTTP调用毁掉了全部异步红利
这个坑让我印象最深。当时把一个接口从Flask迁到FastAPI,压测数据也好看,但上线后P99延迟比压测时高出一大截。查了很久,最后发现接口里用了requests.get()去调用下游服务。
requests是标准同步库,放在async def接口里,等于在所有并发请求共用的那个事件循环里执行阻塞调用。一个请求卡住,后面所有请求排队等它。定位这个问题时,我看代码里的函数签名写的全是async def,很容易误以为整个链路都是异步的。
修复很简单,把requests.get()换成异步客户端,我只举httpx一个例子:
import httpx async def call_downstream(): async with httpx.AsyncClient(timeout=5) as client: resp = await client.get("https://example.com/api") resp.raise_for_status() return resp.json()排查这种问题有一个笨但有效的办法:在代码里全局搜索所有阻塞库的调用,requests、time.sleep、redis.Redis的同步方法、文件读写、subprocess,凡是出现它们的地方,就要Check是不是在async def块里。
5.2 Pydantic嵌套模型在特殊场景下的取舍
Pydantic很爽,但嵌套模型用多了,序列化时间会线性上涨。我有一次给一个列表接口定义了四层嵌套的响应模型,压测发现P99延迟比预期高不少,性能剖析显示大部分时间耗在模型嵌套校验上。
这时候要学会做取舍。响应模型套四层,接口调用方可能根本用不到全部层级。实际处理时我做了两件事:第一,把列表接口的嵌套层数控制在一层,最多两层,深层字段单独提供明细接口;第二,对超大响应体,用response_model_exclude_none=True或response_model_include过滤不必要字段,减小序列化和网络传输的压力。
反过来,请求体验证上用Pydantic嵌套模型是非常值得的,因为它把复杂的入参校验变成声明式行为。但响应体上要克制,嵌套越深,性能开销越大,维护成本也越高。
5.3 测试环境里的event loop冲突
写单元测试时我遇到的另一个坑是event loop冲突。最典型的场景是:测试里同时用pytest-asyncio和httpx的ASGITransport调FastAPI接口,而数据库操作也依赖AsyncSession。如果混用多个event loop,会报"event loop is closed"之类的错误。
我的解决方案是在pytest配置里明确指定asyncio_mode = "auto",并且对涉及数据库的测试统一使用同一个事件循环作用域。如果你用httpx做测试,推荐直接用ASGITransport,不需要真正启动网络端口:
import pytest import httpx async def test_create_order(): transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient( transport=transport, base_url="http://test" ) as client: resp = await client.post( "/orders", json={"product_id": 1, "quantity": 2}, ) assert resp.status_code == 200这样测试跑得快,也不需要额外维护一个测试Uvicorn进程。数据库侧我会用单独配置的测试库,每个用例跑完回滚或清表。一件产品的经验是:把所有测试的fixture集中在conftest.py里管理,不要在单个测试文件里写自己的session逻辑,否则项目测试多了以后巨难维护。
写在最后
这篇文章里的很多选择都有我的主观倾向:比如用SQLAlchemy而不是直接用异步数据库驱动,比如坚持用依赖注入管理会话,比如把压在response_model上的功能边界写清楚。你可以不照搬,但建议理解每个选择背后的取舍。实际使用FastAPI一年多,我最大的体会是:一个框架能帮你提高开发效率,但真正决定接口性能的,仍然是你对异步模型的理解、对数据库访问模式的把控,以及对每一个阻塞点的敏感度。如果读完这篇能让你在新项目里少踩几个坑,我就很满足了。