1. 为什么要单独聊FastAPI的“架构”这两年FastAPI在Python后端圈子里蹿升速度非常快从GitHub上Star的增速和一些大厂内部向异步框架迁移的趋势来看它已经不是“小众玩具”而是真正能扛线上业务的框架之一。很多人学FastAPI只停留在“能用路由写接口、能用Pydantic做校验”这个层面但真正决定项目能不能继续长大、能不能承受复杂业务、能不能在多团队协作里保持清晰的恰恰是它背后的架构设计逻辑而其中最能拉开普通开发者和资深开发者差距的就是依赖注入、后台任务和WebSocket这三块。我先说一个很直观的体会如果你只是写一个几十行的小DemoFastAPI和Flask用起来差别不大无非是路由装饰器和参数声明方式不同。但一旦你的项目开始有数据库连接、有Redis缓存、有配置中心、有用户鉴权、有文件上传、有实时消息推送代码会迅速变得混乱。Flask时代我们靠request.g、靠蓝图、靠一堆手动初始化的全局对象来勉强维持秩序Django时代我们靠中间件和信号量硬撑。而FastAPI给出的是一套更干净的解决方案——依赖注入容器。它让“某个接口需要什么资源”这件事变得显式、可替换、可测试这是架构层面非常关键的进步而不是单纯语法糖。这篇文章我打算围绕三个核心点展开依赖注入的运行机制和工程化用法、后台任务的适用边界和处理方案、WebSocket从握手到推送的完整落地实践。这三个点覆盖了FastAPI开发者日常项目里最高频的架构需求也正好是FastAPI官方文档里“Advanced Usage”和“Concurrency”部分最常被搜索的内容。无论你是在调研技术选型、准备面试还是手上正有一个FastAPI项目需要重构这篇文章都值得花二十分钟仔细读完我会把我在项目中踩过的坑和最后的取舍一并讲清楚。我不会把内容写成官方文档的中文翻译版而是会结合真实项目的演进过程来讲为什么一开始用同步写法没问题后来必须切异步为什么依赖注入不只是“省几行代码”为什么后台任务不是无脑用BackgroundTasks就完事搞清楚这些问题比自己翻一遍文档有用得多。先给这篇文章定个基调适合已经会用FastAPI写基础接口、但想进一步理解框架设计思想并优化项目结构的开发者。如果你还完全没接触过FastAPI建议先照着官方教程跑一个最简单的Demo再回来读体验会更好。2. 依赖注入FastAPI架构中最被低估的设计FastAPI的依赖注入系统其实抄自另一个知名框架——Flask的扩展Flask-Injector和Django的DRF里的某些思路但FastAPI把它做得更顺手、更类型友好。很多初学者只觉得Depends()就是把一个函数作为参数传进去多了一个“依赖”的名头而已这种理解太浅了。依赖注入从架构角度来说解决的核心问题不是“少写几行代码”而是解耦和可测试性。2.1 先搞懂“控制反转”到底反转了什么传统写法的逻辑是“服务内部自己去new一个依赖”比如一个订单服务需要操作数据库直接在函数里创建db create_engine(...)这样写代码最直白但带来的问题是当你想测试这个服务时它一定会真实连接数据库当你想替换数据库连接配置时得去改每一处使用点当你想在中间加一层缓存或日志时得侵入每个调用方。控制反转IoC把“谁负责创建依赖”这件事反转过来服务自己不创建依赖而是声明“我需要一个数据库会话”由框架或容器在合适的时机把数据库会话注入进来。FastAPI的Depends做的就是这件事但它做得很轻量不需要你启动一个什么IoC容器也不需要写一堆配置文件只需要用函数参数的类型声明和默认值描述依赖关系它就能自动处理依赖的实例化和生命周期。我举个日常例子刚开始用FastAPI的开发者大多会这样写from fastapi import FastAPI from sqlalchemy import create_engine app FastAPI() engine create_engine(sqlite:///./test.db) app.get(/users/{user_id}) def get_user(user_id: int): with engine.connect() as conn: result conn.execute(SELECT * FROM users WHERE id ?, (user_id,)) return result.fetchone()这段代码在小项目里完全没问题但想想看如果以后要做性能压测把SQLAlchemy连接池配置、超时时间、回滚逻辑换了你得改动所有接口函数如果要多租户场景每个请求需要根据请求头选择不同的数据库连接这段代码几乎没法扩展。而用依赖注入的方式重构之后from fastapi import FastAPI, Depends from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker app FastAPI() engine create_engine(sqlite:///./test.db) SessionLocal sessionmaker(bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close() app.get(/users/{user_id}) def get_user(user_id: int, dbDepends(get_db)): result db.execute(SELECT * FROM users WHERE id ?, (user_id,)) return result.fetchone()表面上看只是把数据库会话的创建移到了一个函数里但这一步带来的收益是巨大的get_db这个依赖可以被任意多个接口复用测试时只要替换get_db的实现就能在没有真实数据库的情况下测试接口逻辑如果想在依赖里加上请求日志、权限校验、事务处理所有依赖该依赖的接口会自动获得这些能力调用方根本不需要改动。2.2 用yield依赖管理资源生命周期FastAPI依赖注入里最容易被人忽略、但实际工程中非常有用的特性是yield依赖。正常一个依赖函数如果用return返回那么FastAPI在请求结束后不会执行任何清理逻辑但如果依赖函数用yield返回依赖值yield之前的部分是“进入依赖”阶段yield之后的部分是“退出依赖”阶段FastAPI会在请求结束之后自动运行yield后面的清理代码。我最常用的场景就是数据库会话和Redis连接from fastapi import FastAPI, Depends from redis import Redis app FastAPI() def get_redis(): r Redis(hostlocalhost, port6379, decode_responsesTrue) try: yield r finally: r.close() app.get(/health) def health(redis: Redis Depends(get_redis)): redis.set(health_check, ok) return {status: ok}这个写法的妙处在于无论接口执行过程中是否抛出异常finally里的清理代码都会执行不会出现连接泄漏。我在给团队做Code Review时经常看到有人手动在接口函数里try...finally来关闭连接其实在FastAPI里完全可以用yield依赖把这个逻辑收拢到一个地方接口函数本身只关注业务逻辑资源管理交给依赖层这职责划分就清晰了。yield依赖还有一个衍生能力可以做“带状态的依赖”。比如一个依赖要加载用户信息然后在后续逻辑里被多次使用如果每个接口都去查一次数据库显然是浪费这时可以让一个高层依赖先查询用户然后通过yield把数据传递下去。FastAPI的依赖缓存机制保证了在同一个请求内多次调用同一个依赖只会执行一次真正的逻辑这个特性叫“请求级缓存”。2.3 依赖的作用域与请求级缓存控制反转框架里通常会有不同作用域的依赖单例全局唯一、请求级每个请求重新创建、会话级等。FastAPI默认的依赖生命周期是请求级这意味着每个新请求进来依赖函数都会重新执行。但FastAPI在同一个请求内部会缓存同一个依赖函数的结果。我举个例子from fastapi import FastAPI, Depends from typing import Optional app FastAPI() def get_current_user(token: str): # 假设这里通过token解析用户信息 return {username: fuser_{token}} def get_user_from_db(user: dict Depends(get_current_user)): # 模拟数据库查询 print(f查询数据库 user: {user[username]}) return {db_user: user[username], uid: 9527} app.get(/profile) def profile(current_user: dict Depends(get_user_from_db)): return current_user如果请求带上tokenabc那么get_current_user只会执行一次get_user_from_db内部依赖的get_current_user拿到的和接口函数直接依赖的get_current_user是同一个对象。这套机制在工程上非常实用你可以放心地在多个工具函数中声明同一个依赖不必担心重复计算。但要注意一个坑如果依赖函数返回的是一个可变对象而且你在业务逻辑里改了它那么这个修改会影响到同一请求里其他使用该依赖的地方。我在实际项目中就遇到过这种“隐式共享状态”导致的问题某个依赖返回了一个字典A函数往里加了字段B函数拿到的就不再是干净的数据了。所以依赖函数返回的数据尽量设计成不可变对象或者只用于读取。2.4 依赖注入在复杂业务中的架构价值当项目演进到几十上百个接口时依赖注入带来的架构收益会越来越明显。我举一个真实的业务场景一个对外开放的API服务需要三类鉴权方式——App用户Token鉴权、内部服务间调用的API Key鉴权、管理员后台的Session鉴权。如果不用依赖注入每个接口都要复制粘贴一大段鉴权代码而用依赖注入你只需要定义三种依赖函数然后在需要的接口上用Depends(get_current_user)、Depends(get_internal_service)、Depends(get_admin_user)去声明即可。更进一步FastAPI支持依赖嵌套。比如get_current_user需要先依赖get_db获取数据库访问然后去库里查询用户get_permission又依赖get_current_user来获取当前用户并校验权限。这种链式依赖让复杂的业务逻辑可以被拆分成多个小的、可独立测试的函数模块。依赖注入还有一个被忽视的用途——替代中间件做横切逻辑。有时候中间件太“重”因为它会影响所有路由而且拿到上下文比较麻烦。依赖注入则可以在路由级别按需使用。比如某个接口做灰度发布需要读取请求头里的环境标识然后决定走哪套逻辑用依赖注入简直顺滑。从测试角度来说依赖注入最直接的收益是“打桩”很容易。测试时只需要覆盖依赖函数from fastapi.testclient import TestClient import main def get_test_user(): return {username: test_user, role: admin} main.app.dependency_overrides[main.get_current_user] get_test_user client TestClient(main.app) response client.get(/admin/stats) assert response.status_code 200dependency_overrides正是FastAPI为测试准备的官方替换机制这在写单元测试和集成测试时极其舒服不用起Mock服务不用改数据库只要把依赖替换掉接口的行为就完全可控。3. 后台任务把耗时操作移出请求链路Web服务里有一类经典问题请求本身很快但请求触发的某些操作比较耗时比如发邮件、生成缩略图、推送通知、同步第三方数据。如果把这些操作放在请求处理函数里同步执行接口响应时间会暴涨用户会明显感觉到卡顿。后台任务机制就是为这个场景设计的。3.1 BackgroundTasks最简单可靠的后台执行方案FastAPI内置的BackgroundTasks是一个非常轻量的后台任务实现它不会引入新的进程、消息队列或broker而是把任务函数登记在一个列表里等响应发送完毕后由事件循环执行。from fastapi import FastAPI, BackgroundTasks app FastAPI() def send_welcome_email(email: str): # 模拟发邮件耗时操作 import time time.sleep(3) print(f已发送欢迎邮件到 {email}) app.post(/register) def register(username: str, email: str, background_tasks: BackgroundTasks): background_tasks.add_task(send_welcome_email, email) return {message: 注册成功邮件稍后发送}注意background_tasks这个参数不需要你传入任何值FastAPI会自动注入。在响应返回给客户端之后send_welcome_email才真正进入执行。对于发邮件、写审计日志、清理临时文件这类“就算失败了也不影响主流程”的任务BackgroundTasks是最合适的选择。但我要明确一个边界BackgroundTasks只适用于轻量级、可容忍丢失的任务。原因有两个。第一它是在进程内执行的任务队列如果服务进程在任务执行中途崩溃或被重启任务就丢了没有持久化机制。第二它没有重试机制、没有死信队列、没有任务状态跟踪。如果你的任务失败后需要重试或者需要监控任务执行状态BackgroundTasks就不够用了。3.2 异步函数中的后台任务与线程池差异FastAPI支持在后台任务中执行同步函数和异步函数。如果你add_task的是一个async def函数它会作为事件循环里的一个协程任务执行如果你add_task的是一个普通def函数FastAPI会把它扔给默认的线程池执行避免阻塞事件循环。这个细节很关键。很多人写完同步后台任务后发现接口虽然秒回了但其他接口也变慢了原因就是同步后台任务被提交到了AnyIO的线程池如果线程池中的任务本身又要等待I/O且线程池已满新的请求也会被阻塞。所以后台任务的函数体里如果有耗时的I/O操作尽量写成异步版本或者减小任务粒度。我自己常用的模式是后台任务里只发一个信号到消息队列或任务队列真正的重活交给队列的Worker去处理。这样即使任务量突增也不会把Web服务的线程池耗尽。3.3 什么时候必须上Celery/RQ当项目出现以下信号时你就得认真考虑引入专门的任务队列了任务执行时间超过30秒甚至分钟级任务需要重试机制和失败告警任务需要按优先级调度、定时调度多个服务实例会同时消费任务不能容忍重复执行任务执行的结果需要被查询或纳入业务逻辑典型的如视频转码、批量数据导入导出、复杂报表生成、机器学习模型推理等。这种情况下我最常用的是Celery Redis/RabbitMQ的组合。FastAPI和Celery配合的核心思路是FastAPI负责接收HTTP请求并快速返回随后把业务数据处理成一条消息扔进队列Celery Worker独立于FastAPI进程运行监听队列并执行真正的任务。# tasks.py from celery import Celery celery_app Celery(myapp, brokerredis://localhost:6379/0, backendredis://localhost:6379/1) celery_app.task(bindTrue, max_retries3) def process_video(self, video_id: str): try: # 模拟视频处理 print(f正在处理视频 {video_id}) except Exception as e: raise self.retry(exce, countdown60)FastAPI侧只需要把任务描述发给Celeryfrom fastapi import FastAPI, BackgroundTasks from tasks import process_video app FastAPI() app.post(/videos) def upload_video(video_id: str): process_video.delay(video_id) return {message: 视频上传成功处理中}这里process_video.delay(video_id)会立即向Redis写入一条任务消息然后直接返回真正的处理逻辑是异步的而且有重试、有持久化。这就是Celery对比BackgroundTasks的核心优势。但Celery的引入也带来了额外复杂度多一个Worker进程部署、多一个Broker依赖、需要监控任务队列堆积情况。所以我的建议是项目初期不要急着上Celery先用BackgroundTasks或asyncio.create_task顶住等确实出现任务丢失、重试、监控需求时再演进到Celery。3.4 asyncio.create_task与后台任务的坑还有一个常见的替代方案是在异步视图函数里直接使用asyncio.create_taskimport asyncio from fastapi import FastAPI app FastAPI() async def send_notification(user_id: str): await asyncio.sleep(5) print(f已向用户 {user_id} 发送通知) app.post(/users/{user_id}/notify) async def notify(user_id: str): asyncio.create_task(send_notification(user_id)) return {message: 通知任务已创建}这看起来更直接但有一个大坑asyncio.create_task创建的任务不会像BackgroundTasks那样保证在响应发送后才执行而且你很难跟踪任务的生命周期。如果主请求抛异常或者事件循环关闭任务可能永远不执行。更严重的是一旦你使用了多Worker部署多个Worker进程各自的事件循环互不相通任务调度会出现混乱。我把这两种方式的使用场景说清楚BackgroundTasks适合“随请求生命周期”的短任务官方保证响应后再执行代码侵入小asyncio.create_task适合在异步逻辑里临时编排几个协程任务的场景但别把它当任务队列用Celery/RQ适合重任务、长任务、需要可靠投递的任务是架构级方案这三个层级从轻到重你在自己项目里按实际需求选择。我不建议一上来就上Celery过度设计同样会拖垮项目但也不建议所有任务都堆在BackgroundTasks里到时候丢任务丢到怀疑人生。3.5 多进程部署下后台任务重复执行的经典坑用uvicorn main:app --workers 4部署FastAPI时会创建4个独立进程。如果你在启动时用BackgroundTasks注册了一个“启动后执行”的任务它会在4个进程中各执行一次。同样如果每个Worker进程都在内存中维护一个任务队列用户请求被负载均衡到不同Worker上任务也是分散的无法从全局角度管理。解决思路一般有三种一是把所有任务都丢到外部队列Redis/RabbitMQWorker从队列拉任务天然避免重复消费二是如果任务确实只能执行一次使用分布式锁比如Redis的SETNX保证只有一个节点执行三是把任务独立成一个服务由单独的进程调度。这三种方案在真实项目中我都用过最推荐的是第一种成本最低且可扩展性最好。4. WebSocket实战从握手到消息推送WebSocket是建立在TCP之上的全双工通信协议它和HTTP的区别可以理解为HTTP是“一问一答”客户端发请求服务端给响应连接通常就结束了WebSocket则是双方建立一条长连接后任意一方都可以随时向对方推送数据无需每次重新建立连接。这种能力在聊天室、实时通知、协同编辑、股票行情推送、服务端主动事件推送等场景中非常关键。FastAPI对WebSocket的支持在Python后端框架里算是比较成熟的。它在ASGI层面直接支持WebSocket协议不需要像Flask那样通过额外扩展硬套写起来很顺手。4.1 FastAPI最简单的WebSocket实现在FastAPI中定义一个WebSocket端点和定义普通路由非常相似from fastapi import FastAPI, WebSocket, WebSocketDisconnect app FastAPI() app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() try: while True: data await websocket.receive_text() await websocket.send_text(f服务端收到: {data}) except WebSocketDisconnect: print(客户端断开连接)当客户端通过ws://host/ws连接时FastAPI会调用这个函数websocket.accept()表示接受握手请求。之后就是一个无限循环收到客户端消息就回复一条消息。WebSocketDisconnect异常会在客户端断开连接时抛出这是我们做资源清理的入口。这段代码虽然简单但已经涵盖了WebSocket服务端开发最核心的骨架accept握手、receive循环、send推送、disconnect处理。后续所有的复杂应用都是在这个骨架之上扩展的。4.2 连接管理多客户端场景下的广播单客户端连接没什么挑战WebSocket的难点在于多客户端并发连接时的连接管理。比如一个聊天室功能某个用户发送消息服务端要把这条消息广播给房间内所有其他在线用户。要做到这个就得维护一个连接池实时掌握每个连接的状态。一个简单但可靠的方案from fastapi import FastAPI, WebSocket, WebSocketDisconnect from typing import Dict, Set app FastAPI() class ConnectionManager: def __init__(self): self.active_connections: Dict[str, WebSocket] {} async def connect(self, room_id: str, websocket: WebSocket): await websocket.accept() if room_id not in self.active_connections: self.active_connections[room_id] set() self.active_connections[room_id].add(websocket) def disconnect(self, room_id: str, websocket: WebSocket): self.active_connections[room_id].discard(websocket) if not self.active_connections[room_id]: del self.active_connections[room_id] async def broadcast_to_room(self, room_id: str, message: str): if room_id not in self.active_connections: return for connection in list(self.active_connections[room_id]): await connection.send_text(message) manager ConnectionManager() app.websocket(/ws/{room_id}) async def websocket_endpoint(websocket: WebSocket, room_id: str): await manager.connect(room_id, websocket) try: while True: data await websocket.receive_text() await manager.broadcast_to_room(room_id, f用户说: {data}) except WebSocketDisconnect: manager.disconnect(room_id, websocket) await manager.broadcast_to_room(room_id, 有用户离开了房间)ConnectionManager把连接管理的逻辑单独抽了出来用dict把WebSocket连接按房间分组broadcast_to_room遍历连接池逐个发送。这里要注意如果一个连接断开后没有及时从池中移除广播给该连接时会抛异常所以disconnect方法一定要在异常处理中被调到。实际项目里这个ConnectionManager还可以扩展给每个连接绑定用户ID、记录连接创建时间、统计在线人数、支持主动踢人下线等。但无论怎么扩展核心依然是维护一个线程安全、可检索的连接存储结构。4.3 在依赖注入中使用WebSocket和查询参数FastAPI的一大特色是统一了依赖注入体系这个特性在WebSocket端点上同样适用但用法和HTTP路由有一点差别。依赖注入的Depends在WebSocket里也可以使用from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends app FastAPI() async def get_token_from_query(websocket: WebSocket): token websocket.query_params.get(token) if not token: await websocket.close(code4401, reason缺少token) return token app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket, token: str Depends(get_token_from_query)): await websocket.accept() await websocket.send_text(f你的token是: {token}) try: while True: data await websocket.receive_text() await websocket.send_text(data) except WebSocketDisconnect: print(断开)WebSocket的认证常用方式是客户端连接时在URL带上token参数ws://host/ws?tokenxxx服务端在WebSocket函数内部通过websocket.query_params获取并校验然后决定accept还是close。这种方式实现简单但token会出现在URL里安全性相对较低如果对安全要求高更推荐在Sec-WebSocket-Protocol头部里携带认证信息服务端用subprotocol字段做响应。需要注意websocket.close()和accept()的顺序。调用close()后再调用accept()会报错反之亦然。一个连接生命周期内这两个操作只能执行一次所以写代码时要避免重复操作。4.4 WebSocket连接保活与异常处理WebSocket长连接在真实网络环境下会遇到各种问题中间网络设备可能断开空闲连接、客户端网络切换导致断线、服务端重启导致连接失效。因此保活和重连机制是生产环境的必备能力。服务端常用的保活策略有两种发送ping/pong控制帧定期向客户端发送心跳消息。FastAPI的websocket.receive_text()在等待消息时如果客户端长时间不发数据会一直挂起。要检测连接是否存活可以通过设置接收超时import asyncio from fastapi import WebSocket async def receive_with_timeout(websocket: WebSocket, timeout_seconds: int 30): try: return await asyncio.wait_for(websocket.receive_text(), timeouttimeout_seconds) except asyncio.TimeoutError: await websocket.send_text(心跳检查连接仍然活跃) return None这个做法是主动向客户端发送心跳消息客户端收到后如果正常回复说明连接健康。如果客户端不回复服务端可以选择关闭连接让客户端主动重连。我特别想强调一个WebSocket开发中的常见误解不要把状态存到单进程内存里。当你用多Worker部署WebSocket服务时客户端A连接到Worker 1客户端B连接到Worker 2那Worker 1再把消息推给B是推送不到的因为B在另一个进程的连接池中。解决思路是用Redis的Pub/Sub做跨进程消息转发所有Worker订阅同一个Redis频道任意Worker收到WebSocket消息后把消息发布到Redis频道订阅该频道的所有Worker都能收到消息然后在自己管理的连接池中查找目标连接并推送这样通过外部存储打通了多进程间的连接壁垒是从单体WebSocket升级到支持水平扩展的关键一环。4.5 pytest测试WebSocket接口WebSocket接口测试比普通HTTP接口麻烦一些因为涉及连接建立、消息收发、断开连接等多个阶段。FastAPI官方推荐使用TestClient来做WebSocket测试它封装了底层的连接逻辑用起来很方便from fastapi.testclient import TestClient from main import app client TestClient(app) def test_websocket(): with client.websocket_connect(/ws) as websocket: websocket.send_text(hello) data websocket.receive_text() assert data 服务端收到: hellowith块内连接是被保持的退出with块时自动断开。测试用例覆盖的是WebSocket业务逻辑而不需要真正启动一个服务器执行速度快非常适合集成到CI流水线里。对于需要token参数的WebSocketclient.websocket_connect还支持传参with client.websocket_connect(/ws?tokentest_token) as websocket: data websocket.receive_text() assert test_token in data这样写测试的时候服务端依赖的校验逻辑会被真实执行能尽早发现握手阶段的问题。4.6 与前端反向WebSocket应用场景热搜词里出现了一个“Python反向WebSocket”的概念这里很多人会误解。所谓反向WebSocket指的是客户端主动建立WebSocket连接但之后连接的“使用方向”主要是服务端向客户端推送数据。典型应用是内网设备比如打印机、边缘网关、收银机主动连接云端服务端建立一条长连接云端随时通过这条连接下发指令而不是由云端主动去连接内网设备的IP内网设备往往没有公网IP云端根本连不上。这种模式下WebSocket连接的建立仍然是客户端发起的服务端维护连接池等待数据下发。架构上与前面的连接管理类似唯一区别是消息推送主动方是服务端。这套模式在IoT设备管理、企业软件分发、远程运维场景里非常实用。FastAPI的WebSocket支持完全能胜任这类场景。顺带提一句Java的SpringBoot也支持WebSocket用的ServerEndpoint注解其核心逻辑和FastAPI的WebSocket端点其实异曲同工只是语言和框架不同。理解了协议层原理换个语言写WebSocket服务并不难难的是对连接生命周期和消息格式的设计。5. 常见问题与排查技巧实录我把自己和团队在实际项目中遇到过的高频问题汇总成了一张速查表这些问题在网上问答社区被反复提问说明是典型的新手陷阱值得花点时间记下来。问题现象根本原因解决办法接口响应很快但后台任务总是没执行使用了asyncio.create_task又没有正确等待或任务因为异常被事件循环吞掉确认后台任务注册方式在任务函数内加异常捕获并写日志多Worker部署后任务执行了多次BackgroundTasks在每个进程内各执行一次换用Celery等外部队列或用分布式锁保证唯一性连接数据库的接口在高并发下变慢每次请求都创建新的数据库连接没有使用连接池在依赖中使用SQLAlchemy的sessionmaker并正确配置连接池参数websocket.receive_text()一直等待无法向客户端主动推送没有把连接存到连接池服务端缺少主动发送的入口用ConnectionManager保存所有连接在其他接口中调用连接发送消息WebSocket连接总是意外断开网络中间设备空闲超时或者服务端没做心跳保活服务端定期发心跳客户端实现自动重连机制CORS跨域问题在本地联调时出现前端端口和服务端端口不同没有配置CORS中间件在FastAPI中加入CORSMiddleware并配置允许的源列表依赖里的数据库会话在异常后未关闭忘记使用yield依赖或遗漏finally清理改用yield依赖并在finally中执行关闭逻辑部署时连接池报错TimeoutError数据库最大连接数被占满连接池扩容或释放不及时调大连接池上限缩短连接空闲回收时间排查慢查询5.1 CORS配置的“标准答案”本地开发时React/Vue前端跑在http://localhost:5173FastAPI服务跑在http://localhost:8000如果不配置CORS前端连接口必然被浏览器拦截。FastAPI的CORS配置极其简单from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境务必替换成具体来源 allow_credentialsTrue, allow_methods[*], allow_headers[*], )注意allow_origins[*]和allow_credentialsTrue不能同时使用这是浏览器CORS规范的限制。如果你需要携带Cookie或Authorization头就得把allow_origins配置成具体的域名列表不能用通配符。5.2 与SQLAlchemy配合时的性能优化FastAPI项目最常搭配的ORM是SQLAlchemy但如果你没注意到异步驱动和连接池配置性能会大打折扣。推荐直接使用SQLAlchemy 2.0的异步版本from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from fastapi import Depends engine create_async_engine(sqliteaiosqlite:///./test.db, echoTrue, pool_pre_pingTrue) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_db() - AsyncSession: async with AsyncSessionLocal() as session: yield session使用异步数据库驱动如aiosqlite、asyncpg、aiomysql后数据库操作不再阻塞事件循环并发能力会明显提升。pool_pre_pingTrue会在取连接时先做一次轻量探测避免拿到已失效的数据库连接。关于连接池大小经验值是CPU核心数乘以2再加1但要结合数据库的最大连接数限制不能盲目调大。5.3 WebSocket与普通HTTP接口混用时的设计一个服务里既有普通HTTP接口又跑着WebSocket是常态。合理的做法是HTTP接口负责业务操作写入数据、触发逻辑WebSocket只负责“事件推送”。例如用户下单后HTTP接口把订单写入数据库然后向用户的WebSocket连接推送“订单已创建”的通知。在实际项目里我一般会封装一个notify_service内部维护连接池然后在HTTP接口中调用它from fastapi import FastAPI, WebSocket, WebSocketDisconnect app FastAPI() manager ConnectionManager() app.post(/orders) async def create_order(order_id: str): # 业务逻辑保存订单... await manager.broadcast_to_room(order_id, f订单 {order_id} 已创建) return {message: 订单创建成功} app.websocket(/ws/order/{order_id}) async def order_ws(websocket: WebSocket, order_id: str): await manager.connect(order_id, websocket) try: while True: await websocket.receive_text() except WebSocketDisconnect: manager.disconnect(order_id, websocket)这种“HTTP处理业务 WebSocket推送事件”的分工模式比把所有逻辑都塞进WebSocket里清晰得多。如果你让客户端通过WebSocket发创建订单的请求服务端还要处理连接状态与业务状态的耦合问题异常恢复会很麻烦。所以实践中我强烈建议请求走HTTP事件走WebSocket各司其职。5.4 项目目录结构建议关于FastAPI项目的目录结构优化我观察到很多初学者把路由、依赖、模型全塞在一个main.py里几百行还好上千行后维护效率直线下降。这里贴一个我在中等项目中验证有效的分层结构app/ ├── main.py ├── core/ │ ├── config.py │ ├── security.py │ └── deps.py ├── api/ │ ├── v1/ │ │ ├── endpoints/ │ │ │ ├── users.py │ │ │ └── orders.py │ │ └── router.py ├── models/ │ ├── user.py │ └── order.py ├── schemas/ │ ├── user.py │ └── order.py ├── services/ │ ├── order_service.py │ └── notify_service.py ├── workers/ │ └── tasks.py └── tests/ └── test_websocket.pycore放配置和安全工具api放路由层只做参数接收和响应封装services放业务逻辑models和schemas分别是ORM模型和Pydantic数据校验模型workers专门放后台任务和Celery任务。依赖注入函数统一放core/deps.py方便管理和测试。按这个结构组织代码项目到几万行代码依然能保持清晰。最后再分享一个我个人的经验FastAPI的上手门槛确实很低但真正用好它需要建立“异步优先、依赖驱动、任务分层”的思维方式。动手写项目之前先花半小时把所有路由和依赖关系画个草图想想哪些逻辑可以抽成依赖哪些操作应该异步化哪些消息需要走WebSocket推送。架构设计不是写代码之后才考虑的补救措施而是写代码之前就应该形成的骨架。你按照这个思路去组织FastAPI项目后续的维护和扩展会轻松很多。
