多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

FastAPI 实战指南:从架构设计、异步性能优化到生产部署

FastAPI 实战指南:从架构设计、异步性能优化到生产部署 很多人问我后端选型到底看什么。我的观点很朴素如果业务是重 I/O、轻计算比如读写数据库、对接第三方服务、给前端或小程序提供接口那 FastAPI 基本能让你少写三分之一的胶水代码还不牺牲响应速度。它原生支持 async/await类型注解就是校验和文档来源依赖注入帮你在不同路由里复用同一套逻辑你只需要把精力聚焦在业务本身。这篇内容是我把一套真实项目从零搭到上线时的方法和踩坑记录适合后端开发、全栈工程师也适合正在评估新项目技术选型但没时间把所有框架都试一遍的人。1. 为什么选 FastAPI我只关心三件事1.1 异步原生你的服务不是慢是被同步代码卡住了很多人一上来就问“FastAPI 性能是不是比 Flask 好”其实这问题本身是伪命题。Flask 是同步框架每个请求的到来会让线程池里的一个线程被占住线程等待数据库、等待外部接口响应时它不干活但也不能被复用。就像奶茶店里每个顾客身后都跟着一个贴身服务员顾客只是盯着奶茶发呆服务员也得陪着。线程多到一定程度内存和上下文切换开销就上来了。FastAPI 的异步模型走的是另一条路当接口函数被你写成async def它在碰到等待数据库查询、等待 HTTP 响应这类操作时会把当前任务的执行状态挂起把事件循环让给别人。这等于一个服务员可以同时接待一百个顾客他给你下完单就去招呼别人等你的奶茶好了他再回来把奶茶端给你。结果就是单位时间内能承接的请求数量高出很多尤其是当请求里有大量等待操作时提升不是百分之几十而是好几倍。需要说明的是FastAPI 也允许你写def接口。如果你用def它会把函数放进线程池去执行保证不会阻塞事件循环。所以你会发现 FastAPI 其实是一个“同步异步双轨”的框架。问题是很多初学的人全都写def或者全都写async def却不知道两者分别适用什么场景性能自然出不来。我的经验是如果是简单读缓存、数据校验、CPU 密集计算用def反而更稳如果是查数据库、调外部 API、读文件优先用async def再配合异步驱动才有效果。1.2 类型提示与 Pydantic文档和校验不是额外的事是顺手长出来的现代 API 开发里最琐碎的环节之一就是参数校验。手写if id not in request的代码不仅容易漏还会让文档失联。FastAPI 之所以能“现代”核心在于它把 Python 的类型标注变成了运行时约束。你写一个函数参数q: str Query(defaultNone, max_length50)它就知道这个参数应该怎么校验校验不过会直接返回 422你的业务代码不用碰脏数据。同时只要你定义了response_modelFastAPI 会自动对响应数据做序列化、过滤多余字段、生成 OpenAPI 文档。前端小伙伴不需要你额外整理接口文档旁边跑一个/docs就能直接试接口。我第一次带团队用 FastAPI 时最明显的感受就是“接口对账”这件事几乎退出了日常协作——后端定义的字段和前端看到的一定是一致的因为它们来自同一套 Pydantic 模型。这背后有个容易被忽略的点Pydantic 的校验是可以被反复调用的但如果你把同一份数据在多处手动传给dict()或json.dumps()就会丢失类型约束和字段过滤能力。正确做法是让 Pydantic 模型做唯一出口。我也踩过坑为了“性能优化”试图用model_dump()绕开部分字段校验结果前端收到一个多字段的 JSON联调半天才发现。其实 Pydantic v2 的性能已经非常可观正常业务场景下不需要你做这种微优化。2. 项目结构别等到要改第二版才开始重构2.1 一套拿来就能用的目录分层很多 FastAPI 教程只教你怎么在一个文件里写三个接口。真实项目绝不能这么干。我在实际项目里用的是下面这套结构它不算复杂但足够支撑一个中大型 API 持续迭代。app/ ├── api/ │ ├── v1/ │ │ ├── __init__.py │ │ ├── routers/ │ │ │ ├── health.py │ │ │ ├── links.py │ │ │ └── users.py │ │ └── dependencies.py ├── core/ │ ├── config.py │ ├── logging.py │ └── security.py ├── models/ │ ├── base.py │ └── link.py ├── schemas/ │ ├── __init__.py │ └── link.py ├── services/ │ ├── link_service.py │ └── stats_service.py ├── repositories/ │ └── link_repo.py ├── main.py └── worker.py有人会说这不是把简单问题复杂化了吗我的回答是目录结构不是给电脑看的是给三个月后的你和同事看的。routers只处理 HTTP 参数和响应services放业务规则repositories封装数据库操作。这样分层之后换数据库驱动不影响路由层改权限逻辑不影响业务层写单元测试时也能轻松 mock 掉某一层。否则所有逻辑堆在路由函数里上一个需求还能改两个需求并行开发时就开始互相打架了。2.2 依赖注入的工程意义数据库会话和鉴权代码不再到处复制FastAPI 的Depends不是炫技是拿来干活的。以数据库会话为例如果每个路由都手动await session make_session()、finally: close()一旦漏了 close连接池就会被耗尽。更难受的是你很难一次性在几十个路由上统一加超时配置或埋点。用依赖注入可以这样定义get_db依赖然后路由函数里写db: AsyncSession Depends(get_db)。FastAPI 会在每个请求开始时调用依赖结束时会帮你清理。你再写一个get_current_user把它放在需要登录的接口上鉴权逻辑就从业务代码里彻底剥离了。同一个依赖在多个路由里被复用改一处就是改全局这才是工程上的“一劳永逸”。新手最容易犯的错是过度依赖注入把所有参数都包一层Depends。其实注入的对象应该是“跨路由复用的资源”而不是每个路由特有的临时参数。如果你发现自己在一个路由里注入了七个依赖那很可能说明这个路由承担的职责太多了该拆函数了。2.3 路由组织的两个经验文件别按 CRUD 拆层与层之间别互相引用做 API 目录设计时我最开始也犯过“一个资源一个文件”的错误比如users.py里同时放用户注册、登录、个人信息、修改密码。结果呢上百行不说路由的权限前缀也不一样一个文件里混了Public、Auth、Admin三类接口每次打开都要往上翻半天。后来我改成按“接口能力”来拆文件。以链接服务为例links.py只放短链接的创建与跳转stats.py放统计查询users.py放账号相关。这个尺度没有标准答案但判断标准很简单当你要加一个新接口时你能不能在第一秒说出该加到哪个文件如果能说明结构是健康的。层与层之间的引用更是重灾区。我见过models层直接 import 路由层然后路由层又 importmodels的循环引用现场。规矩只有一条依赖方向必须是想清楚后不让它反过来的那种api可以依赖servicesservices可以依赖repositories但反向不要做。真出现循环引用不要慌先看是不是有公共对象被放到了core或schemas里绝大多数循环引用是“共享的常量放在哪一层”的问题而不是架构问题。3. 高性能的关键动作把异步优势真正用起来3.1 数据库访问优先异步驱动连接池也别开得太大FastAPI 本身再快如果数据库访问是同步的性能瓶颈就会立刻转移到数据库连接上。每个请求都新建一个连接握手、鉴权、断开这些开销远比你想象的大。我在项目里用的组合是 SQLAlchemy 2.0 的create_async_engineasyncpg驱动。你可能会问为什么不是直接 psycopg3 或者异步 SQLAlchemy 核心因为真实项目里的业务复杂度需要 ORM 来减少重复 SQLSQLAlchemy 2.0 的异步支持已经相当成熟配合 alembic 做迁移也很顺。一个常见的误区是“连接池越大越好”。实际上 asyncpg 的默认pool_size只有 10有些团队一上来就调到 50、100。数据库服务端能同时处理的连接数是有限的连接过多会导致锁竞争和内存占用上升尤其在高并发场景下连接数比请求数先打满不是好事。我的建议是先从 10 到 20 开始压测观察数据库 CPU 和连接等待时间再决定是否上调。真正压垮数据库的不是请求量而是同时涌进来的连接数。3.2 调用外部 HTTP别在 async 路由里写 requests.get这是我在评审代码时最常指出的一个问题。有人在async def路由里写requests.get(...)问就是“之前一直这么写的”。你要理解requests是同步库它发起请求后当前线程会一直阻塞到响应返回。这在 FastAPI 的异步事件循环里是一个大忌它相当于你包了一个光滑的视频播放器但电源线却缠在了一起播放器再好也使不上劲。正确做法是用httpx.AsyncClient并且不要每个请求都新建一个客户端。客户端连接池和连接复用是性能提升的关键就像你不会每次拿快递都重新申请一个快递柜。你可以把httpx.AsyncClient挂到app.state上或者放到依赖里复用。下面是一个最小化示例import httpx from fastapi import FastAPI app FastAPI() app.on_event(startup) async def startup(): app.state.client httpx.AsyncClient(timeout10) app.on_event(shutdown) async def shutdown(): await app.state.client.aclose() app.get(/do-something) async def do_something(): resp await app.state.client.get(https://example.com/api) return resp.json()app.on_event在新版本里可能不再是推荐写法但思路是一样的生命周期里共享一个客户端关闭时释放资源。这一步做完你会发现外部调用的并发能力上了一个台阶。3.3 序列化与中间件别让每次响应都做重复计算FastAPI 返回的 JSON 序列化默认用json模块。大多数场景够用但如果你对响应速度有执念可以换用orjson。Pydantic 本身也支持通过model_config配置序列化器能省掉不少字典转换的隐式开销。另一个很容易被忽视的优化点是中间件顺序。比如CORSMiddleware、TrustedHostMiddleware这些中间件应该在路由逻辑前面还是后面中间件是一个洋葱模型越外层包得越多。如果把重计算放进中间件比如在中间件里做全量日志记录请求响应时间就会被明显拉长。我的经验是中间件里只放跨领域关注点比如 CORS、日志请求 ID、超时控制业务上的数据加工一定要放在路由和服务层里这样你可以针对单接口做优化而不影响全局。对于热点数据不要每次都查数据库。我一般用 Redis 做缓存缓存读取是纯内存操作异步查一次通常在毫秒级。但缓存不是银弹要注意缓存穿透和缓存击穿。一个最简单的防线是设置合理的 TTL 并加入“空值缓存”如果数据库里没有这个 key就把空结果缓存几十秒防止恶意请求反复打库。3.4 压测一个真实接口瓶颈通常不在 FastAPI 里我随手写一个简单的异步接口压测过机器一般的话跑 2000-4000 QPS 并不难。真正让 QPS 掉到几百的往往是下面几个地方数据库连接池太小、外部调用没有超时、日志处理是同步的、事件循环里做了 CPU 密集运算。一旦遇到瓶颈不要急着优化框架先开py-spy dump看每个进程都在干什么或者接入 OpenTelemetry把每个请求的时间分布采集出来。性能优化的原则永远是先测量再动手而不是靠感觉改代码。4. 实战用 FastAPI 写一个短链接服务4.1 需求与接口设计为了让你看个完整示例我挑了一个特别能体现 FastAPI 优势的服务短链接。需求不复杂提交一个长网址生成短码并返回访问短码重定向到原始网址查询短码的访问次数接口设计很直接POST /shorten body: {url: https://...} 响应: {code: abc123, short_url: /abc123} GET /{code} 重定向 302 到原始网址 GET /stats/{code} 返回访问次数短码我建议用secrets模块生成不要用 uuid 的完整串会很长。可以用 base62 编码压缩一下或者直接用nanoid库。4.2 核心实现异步路由 异步数据库会话主程序文件大致是这样的from fastapi import FastAPI, Depends, HTTPException, status from fastapi.responses import RedirectResponse from sqlalchemy.ext.asyncio import AsyncSession from api.v1.dependencies import get_db from repositories.link_repo import LinkRepo from services.link_service import LinkService from schemas.link import ShortenRequest, ShortenResponse app FastAPI(title短链接服务, version1.0.0) app.post(/shorten, response_modelShortenResponse, status_codestatus.HTTP_201_CREATED) async def shorten(req: ShortenRequest, db: AsyncSession Depends(get_db)): service LinkService(db) record await service.create_short_link(req.url) return ShortenResponse(coderecord.code, short_urlf/{record.code}) app.get(/{code}, status_codestatus.HTTP_307_TEMPORARY_REDIRECT) async def redirect(code: str, db: AsyncSession Depends(get_db)): repo LinkRepo(db) record await repo.get_by_code(code) if record is None: raise HTTPException(status_code404, detailshort link not found) await repo.increment_views(record.id) return RedirectResponse(record.original_url, status_code307)LinkRepo里只用异步查询比如from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession class LinkRepo: def __init__(self, session: AsyncSession): self.session session async def get_by_code(self, code: str): result await self.session.execute( select(LinkModel).where(LinkModel.code code) ) return result.scalar_one_or_none() async def increment_views(self, link_id: int): await self.session.execute( update(LinkModel) .where(LinkModel.id link_id) .values(viewsLinkModel.views 1) ) await self.session.commit()注意get_by_code用scalar_one_or_none()返回None时路由层就能判断并返回 404。别小看这个细节如果你用了first()再判断语义就不够清晰了。提交时机也要把握好读接口只负责读但跳转接口里更新访问量是写操作所以我在跳转时只做 update 和 commit。业务虽然简单但分层后你能明显看到每个函数都可以单独测。4.3 压测与调优从 700 QPS 到 3500 QPS 的过程我用locust对这个短链接服务做了一次小规模压测机器是 4 核 8G数据库是本机 PostgreSQL。第一轮压测时 QPS 只有 700 出头我本以为异步接口可以承受更高一看性能数据发现瓶颈在数据库连接池默认pool_size5当 200 个并发进来时大量请求都在等连接。把连接池调到 20QPS 到了 1800 左右。接着发现访问量更新和跳转耦合在同一次请求里读接口也带着写操作竞争加重。我便用 Redis 缓存长网址映射命中缓存时直接跳转不再查数据库只在最终统计时异步落库。这轮调优后稳定在了 3500 左右。这个结果不是绝对的但它说明一件事FastAPI 能跑多快取决于你喂给它的下游资源是否够快。框架本身很少是瓶颈瓶颈通常在连接、锁、日志和下游调用上。5. 部署阶段的坑日志、并发模型和健康检查5.1 uvicorn 日志丢失默认配置会吃掉你的调试线索不少人在本地用 uvicorn 跑得好好的部署之后就发现日志时有时无。uviron 默认使用 Python 的logging模块但如果你在代码里用了别的日志库比如 loguru或者没有为uvicorn.error和uvicorn.access配置 handler日志就可能被吞掉。我见过最诡异的现象是接口能通但访问日志完全没有排错时只能靠猜。我的建议是不要在多个地方各写一套日志。把所有日志统一交给 Python logging并配置两个 handler一个输出到 stdout一个输出到文件。部署时再用日志采集方案把 stdout 收走。如果你用 loguru可以通过一个拦截器把 uvicorn 的日志重定向到 loguru否则到线上看到的日志格式会非常混乱。另一个细节是log_level要显式设置比如uvicorn main:app --host 0.0.0.0 --port 8000 --log-level info默认的 warning 会把很多有用的请求日志过滤掉。5.2 Worker 数量与并发模型不是越多越好uvicorn 单进程能撑起大量异步请求但 Python 进程有 GIL单进程的 CPU 利用率仍然受限。部署时常用的做法是用 Gunicorn 作为进程管理器以 uvicorn worker 的形式跑多个进程。生产环境我一般从2 * CPU 核心数 1开始试。对 4 核机器就是 9 个 worker。但这不是一个铁律如果服务里大量使用异步 I/O进程数小一点也够如果服务里有较多 CPU 密集逻辑进程数稍微多一点更好。最高效的办法是压测在你需要的并发目标下看 CPU 和内存占用找到拐点而不是堆数字。另外要说一个容易搞混的概念uvicorn 的--workers是启动多个进程不是线程。线程池只在def类型的接口里使用默认是 40 个线程。如果你的业务里有需要同步库的第三方调用可以适当调大--limit-concurrency和线程数但本质上你把同步调用塞进线程池只是“不阻塞事件循环”不是让系统并发能力变得无限大。5.3 容器化部署健康检查必须做优雅关闭必须做容器部署时的 Dockerfile 不需要把整个虚拟环境塞进去用多阶段构建会小很多。运行时阶段用python:3.12-slim安装依赖后拷贝代码然后区分启动命令。FROM python:3.12-slim as builder WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir --prefix/install -r requirements.txt FROM python:3.12-slim WORKDIR /app COPY --frombuilder /install /usr/local COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000, --workers, 4]容器里还要加一个/healthz接口用来给 Kubernetes 或负载均衡做探活。健康检查里应该包含数据库连通性检查但注意不能每次都做复杂查询。比如每 10 秒探一次你可以只SELECT 1。如果数据库不可用探活失败会触发重启但探活本身也不要拖太久超时设置短一点。优雅关闭容易被忽略。发布新版本时旧容器一旦收到 SIGTERM 信号就立刻停止正在处理的请求就会中断。在 FastAPI 里可以用lifespan配合信号注册等待当前请求处理完再退出。不用自己写复杂的信号处理至少保证部署工具在停止前有terminationGracePeriod让服务有时间处理完存量请求。6. 常见问题与排查技巧实录6.1 调用外部 API 时报鉴权、超时、连接重置怎么办表格整理几个我在项目里真实遇到的报错报错信息可能原因排查方向unexpected status 401 unauthorized: incorrect api key providedAPI Key 写错、复制时带了空行、Key 已失效或权限不足检查环境变量前后空格到对应平台控制台重新生成并确认权限范围api error: 400 this models maximum context length is ... tokens请求内容太长超出模型上下文窗口做摘要、分片或换用支持更长上下文的模型核对 token 计算口径connection dropped (econnreset)外部服务主动断开连接或本机到目标网络不稳定增加重试逻辑调大超时时间确认是否触发了服务端限流permission denied while trying to connect to the docker api当前用户没有 Docker socket 权限将用户加入 docker 组或用环境变量指向正确的 Docker host请求偶发超时连接池耗尽、DNS 解析慢、服务端资源不足看连接数、线程数是否打满做慢查询日志考虑连接预热这些错误和 FastAPI 本身无关但往往都发生在部署后的第一次联调里。排查时我的习惯是先看请求具体带出去的 Header 和 Body再在目标服务端看日志。很多“鉴权失败”根本不是密钥问题而是你在代码里把密钥转成了全小写或者密钥变量在 CI 环境里没有正确注入。6.2 写接口测试的几个细节依赖覆盖别遗漏FastAPI 内置的TestClient是基于httpx的同步客户端用起来很顺手。但我要提醒几个坑第一测试数据库要用独立库不能复用开发库。最省事的方式是在测试文件里创建临时库用完销毁或者用内存型数据库。第二写测试时如果你用到了Depends(get_db)需要会覆盖依赖app.dependency_overrides[get_db] override_get_db这样路由里的数据库会话就会换成测试会话不会污染线上数据。第三TestClient内部会触发事件循环如果你直接在 pytest 的async def测试函数里调用它会容易出现事件循环冲突。简单办法就是测试函数写成普通的def。等后来需要测更复杂的异步编排再上pytest-asyncio和httpx.AsyncClient。6.3 安全与合规基线密钥、限流、跨域FastAPI 可以很快但安全配置必须从一开始就做。最基本的三件事密钥别写进代码里。用.env配合pydantic-settings读取.env加入.gitignore。环境变量里即使泄露也比代码仓库泄露要好处理。加限流。可以用slowapi也可以自己用 Redis 记 count。简单限流规则是“每 IP 每分钟最多请求 N 次”更高一点的需求按用户 ID 做限流。不加限流的结果就是偶尔一次流量抖动就能把后端打崩。设置 CORS。如果你的接口要提供给浏览器调用CORSMiddleware只配置必要的域名不要图省事写allow_origins[*]。同时把TrustedHostMiddleware加上防止恶意 Host 头投毒。另外不要把敏感数据打进日志。有些日志库会默认记录请求头如果请求头里有 Authorization那你的密钥和 token 就会以明文形式躺在日志系统里。要么在 log filter 里过滤要么在日志接入层做脱敏这个成本很低但出事之后的成本很高。写在最后的一个个人体会我在实际使用中最受用的一点是FastAPI 给你的不只是“框架”而是一整套约束类型、依赖、请求生命周期都有一条明确的路径。你顺着它的路子走代码会越来越顺你非得绕开它去用同步库、全局变量、裸 SQL 拼字符串那再快的框架也帮不了你。如果你正准备做一个新项目我会建议你从一个小接口、一条数据库查询开始把异步链路和依赖注入跑通然后试着接外部 API、写缓存、加日志。每一步都亲自动手压一压再去看那些“性能调优”的经验你才能真正理解它们为什么存在。踩过几次坑之后你会喜欢上这套工作流的。
返回列表