文章目录
- 前言
- 一、阶段学习目标（半天速成）
- 二、核心一：BackgroundTasks 后台任务（解耦耗时操作）
- 2\.1 核心原理
- 2\.2 基础实战：简单后台任务
- 2\.3 多任务叠加 \+ 异步任务支持
- 2\.4 关键避坑点（生产必看）
- 三、核心二：WebSocket 实时通信（进阶核心）
- 3\.1 WebSocket 优势
- 3\.2 基础实战：一对一消息通信
- 3\.3 进阶实战：全员广播聊天室
- 四、高级特性一：流式响应与文件下载
- 4\.1 流式文本响应（实时输出）
- 4\.2 静态文件托管 \+ 文件下载
- 五、高级特性二：精细化参数与响应控制
- 5\.1 参数别名（前端传参与后端字段不一致）
- 5\.2 自定义响应头、状态码
- 六、BackgroundTasks vs Celery 选型（生产核心）
- 七、新手高频避坑总结
前言
这是 FastAPI 零基础系列最后一个进阶阶段，前面我们搞定了接口开发、数据校验、数据库、鉴权、工程化架构、中间件跨域。本阶段只用半天时间，补齐项目高阶能力短板，让你的项目从“普通接口服务”升级为“支持异步任务、实时通信、高性能”的完整后端服务。
本阶段核心解决三大业务痛点：
1.接口阻塞问题：发送邮件、日志记录、文件预处理、数据统计等耗时操作，卡住接口响应；
2.无法实时通信：传统HTTP短链接无法实现消息推送、在线聊天、实时通知、日志推送；
3.高级功能缺失：静态文件托管、流式响应、大文件下载、参数别名等生产常用特性。
学完本节，你的 FastAPI 项目将具备企业级完整能力，可直接用于后台管理、实时系统、消息服务、自动化任务项目。
一、阶段学习目标（半天速成）
1. 精通 FastAPIBackgroundTasks 后台任务，掌握同步/异步任务、多任务叠加、使用场景与避坑点；
2. 掌握WebSocket 实时通信，实现一对一实时消息、在线连接管理、服务端主动推送；
3. 掌握流式响应、文件下载、静态资源托管高级特性；
4. 学会参数别名、忽略字段、自定义响应头、状态码精细化控制；
5. 区分 BackgroundTasks 与 Celery 重型任务队列的适用场景；
6. 整合所有进阶特性，完成最终完整版项目能力闭环。
二、核心一：BackgroundTasks 后台任务（解耦耗时操作）
2.1 核心原理
传统接口：耗时任务执行完毕 → 才返回响应，用户需要长时间等待。
后台任务：接口先返回响应，客户端立即收到结果，耗时任务在后台异步执行，完全不阻塞主请求流程。
适用场景：发送短信/邮件、操作日志记录、数据统计、文件异步处理、消息推送、缓存更新等无需即时返回的操作。
2.2 基础实战：简单后台任务
fromfastapiimportFastAPI,BackgroundTasksimporttime app=FastAPI(title="后台任务实战")# 定义耗时后台任务（普通同步函数）defsend_welcome_email(email:str):# 模拟邮件发送耗时2stime.sleep(2)print(f"【后台任务执行成功】向{email}发送欢迎邮件")@app.post("/register",summary="用户注册（后台发邮件）")defregister(email:str,background_tasks:BackgroundTasks):# 添加后台任务，接口响应后自动执行background_tasks.add_task(send_welcome_email,email)# 立即返回响应，无需等待邮件发送完成return{"code":200,"msg":"注册成功，邮件即将发送"}执行效果：接口瞬间返回结果，2秒后控制台打印邮件发送日志，完全不阻塞请求。
2.3 多任务叠加 + 异步任务支持
BackgroundTasks 同时支持同步函数、async异步函数，且一个接口可叠加多个后台任务。
importasyncio# 异步后台任务asyncdefasync_write_log(username:str):awaitasyncio.sleep(1.5)print(f"【异步日志】用户{username}完成注册操作")# 同步耗时任务defsync_statistics():time.sleep(1)print("【数据统计】更新用户注册数据完成")@app.post("/register/multi",summary="注册+多后台任务")asyncdefregister_multi(username:str,email:str,background_tasks:BackgroundTasks):# 叠加多个后台任务background_tasks.add_task(send_welcome_email,email)background_tasks.add_task(async_write_log,username)background_tasks.add_task(sync_statistics)return{"msg":"注册成功，后台任务异步执行中"}2.4 关键避坑点（生产必看）
1. ❌BackgroundTasks 依附请求生命周期：服务重启、请求异常中断，后台任务会丢失，不适合长耗时、高可靠任务；
2. ❌ 不能传递数据库会话等瞬时资源（请求结束后会话关闭，任务报错）；
3. ✅ 轻量短时任务（10s内）用 BackgroundTasks，长耗时/高可靠任务用 Celery；
4. ✅ 任务函数参数仅传递基础数据（字符串、数字、字典），禁止传递请求对象、DB会话。
三、核心二：WebSocket 实时通信（进阶核心）
3.1 WebSocket 优势
HTTP 是短链接：一次请求一次响应，无法主动推送数据；
WebSocket 是长链接：客户端与服务端建立永久连接，双向实时通信，适合消息推送、在线聊天、实时日志、设备监控、在线人数统计。
3.2 基础实战：一对一消息通信
fromfastapiimportWebSocket,WebSocketDisconnect# 存储在线连接（简易在线管理）active_connections:list[WebSocket]=[]@app.websocket("/ws/chat")asyncdefwebsocket_chat(websocket:WebSocket):# 接受客户端连接awaitwebsocket.accept()# 加入在线列表active_connections.append(websocket)try:whileTrue:# 接收客户端消息data=awaitwebsocket.receive_text()print(f"收到客户端消息：{data}")# 服务端主动回复消息awaitwebsocket.send_text(f"服务端已收到：{data}，时间：{time.time():.0f}")exceptWebSocketDisconnect:# 客户端断开连接，移除在线连接active_connections.remove(websocket)print("客户端断开连接")调试方式：访问 /docs，找到 websocket 接口，直接在线连接、发送消息，无需额外前端页面。
3.3 进阶实战：全员广播聊天室
实现一个客户端发消息，所有在线客户端实时接收的广播功能，适配多人聊天场景。
classConnectionManager:def__init__(self):self.active_connections:list[WebSocket]=[]asyncdefconnect(self,websocket:WebSocket):awaitwebsocket.accept()self.active_connections.append(websocket)defdisconnect(self,websocket:WebSocket):self.active_connections.remove(websocket)asyncdefbroadcast(self,message:str):# 全员广播消息forconnectioninself.active_connections:awaitconnection.send_text(message)# 全局连接管理器manager=ConnectionManager()@app.websocket("/ws/broadcast")asyncdefwebsocket_broadcast(websocket:WebSocket):awaitmanager.connect(websocket)try:whileTrue:data=awaitwebsocket.receive_text()# 广播给所有在线用户awaitmanager.broadcast(f"【全员消息】{data}")exceptWebSocketDisconnect:manager.disconnect(websocket)四、高级特性一：流式响应与文件下载
4.1 流式文本响应（实时输出）
适用于AI问答、日志实时打印、大文本分片输出，避免一次性加载全部数据占用内存。
fromfastapi.responsesimportStreamingResponseimportiodefstream_text():foriinrange(1,6):yieldf"实时分片数据{i}\n"time.sleep(0.5)@app.get("/stream",summary="流式响应")asyncdefget_stream():returnStreamingResponse(stream_text(),media_type="text/plain")4.2 静态文件托管 + 文件下载
FastAPI 原生支持静态资源托管，可直接搭建简易文件服务、后台静态页面。
fromfastapi.staticfilesimportStaticFilesfromfastapi.responsesimportFileResponse# 托管static目录下所有静态资源（图片、html、js、css）app.mount("/static",StaticFiles(directory="static"),name="static")# 单文件下载接口@app.get("/download",summary="文件下载")asyncdefdownload_file():returnFileResponse("test.txt",filename="下载文件.txt")五、高级特性二：精细化参数与响应控制
5.1 参数别名（前端传参与后端字段不一致）
frompydanticimportBaseModel,FieldclassUserQuery(BaseModel):# 前端传 user_name，后端映射为 usernameusername:str=Field(None,alias="user_name")age:int=Field(None,ge=0)@app.get("/query/user")asyncdefquery_user(params:UserQuery):returnparams.model_dump(by_alias=True)5.2 自定义响应头、状态码
fromfastapiimportResponse@app.get("/custom/resp")asyncdefcustom_response(response:Response):# 自定义状态码response.status_code=201# 自定义响应头response.headers["X-App-Version"]="1.0.0"response.headers["X-Server-Name"]="FastAPI-Pro"return{"msg":"自定义响应完成"}六、BackgroundTasks vs Celery 选型（生产核心）
| 特性 | BackgroundTasks | Celery 任务队列 |
|---|---|---|
| 架构 | 内置轻量，无额外服务 | 独立任务队列，依赖Redis/MQ |
| 可靠性 | 低，服务重启任务丢失 | 高，支持重试、持久化、延迟任务 |
| 耗时场景 | 短时任务（10s内） | 长耗时、定时任务、海量任务 |
| 适用项目 | 小型项目、简单异步场景 | 中大型生产项目、分布式系统 |
七、新手高频避坑总结
1. ❌ 后台任务使用请求内瞬时资源（DB会话、请求对象），导致报错；
2. ❌ 依赖 BackgroundTasks 执行长耗时核心业务，任务丢失造成数据异常；
3. ❌ WebSocket 不做异常捕获，客户端断开导致服务报错；
4. ❌ 静态文件目录不存在，导致服务启动失败；
5. ✅ 轻量异步用内置后台任务，核心异步任务务必上 Celery；
6. ✅ WebSocket 必须统一管理在线连接，规范处理断开异常；
7. ✅ 流式响应优先用于大文件、实时输出场景，减少内存占用。
