文章目录前言一、阶段学习目标半天速成二、核心一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 appFastAPI(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长耗时/高可靠任务用 Celery4. ✅ 任务函数参数仅传递基础数据字符串、数字、字典禁止传递请求对象、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:# 接收客户端消息dataawaitwebsocket.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)# 全局连接管理器managerConnectionManager()app.websocket(/ws/broadcast)asyncdefwebsocket_broadcast(websocket:WebSocket):awaitmanager.connect(websocket)try:whileTrue:dataawaitwebsocket.receive_text()# 广播给所有在线用户awaitmanager.broadcast(f【全员消息】{data})exceptWebSocketDisconnect:manager.disconnect(websocket)四、高级特性一流式响应与文件下载4.1 流式文本响应实时输出适用于AI问答、日志实时打印、大文本分片输出避免一次性加载全部数据占用内存。fromfastapi.responsesimportStreamingResponseimportiodefstream_text():foriinrange(1,6):yieldf实时分片数据{i}\ntime.sleep(0.5)app.get(/stream,summary流式响应)asyncdefget_stream():returnStreamingResponse(stream_text(),media_typetext/plain)4.2 静态文件托管 文件下载FastAPI 原生支持静态资源托管可直接搭建简易文件服务、后台静态页面。fromfastapi.staticfilesimportStaticFilesfromfastapi.responsesimportFileResponse# 托管static目录下所有静态资源图片、html、js、cssapp.mount(/static,StaticFiles(directorystatic),namestatic)# 单文件下载接口app.get(/download,summary文件下载)asyncdefdownload_file():returnFileResponse(test.txt,filename下载文件.txt)五、高级特性二精细化参数与响应控制5.1 参数别名前端传参与后端字段不一致frompydanticimportBaseModel,FieldclassUserQuery(BaseModel):# 前端传 user_name后端映射为 usernameusername:strField(None,aliasuser_name)age:intField(None,ge0)app.get(/query/user)asyncdefquery_user(params:UserQuery):returnparams.model_dump(by_aliasTrue)5.2 自定义响应头、状态码fromfastapiimportResponseapp.get(/custom/resp)asyncdefcustom_response(response:Response):# 自定义状态码response.status_code201# 自定义响应头response.headers[X-App-Version]1.0.0response.headers[X-Server-Name]FastAPI-Proreturn{msg:自定义响应完成}六、BackgroundTasks vs Celery 选型生产核心特性BackgroundTasksCelery 任务队列架构内置轻量无额外服务独立任务队列依赖Redis/MQ可靠性低服务重启任务丢失高支持重试、持久化、延迟任务耗时场景短时任务10s内长耗时、定时任务、海量任务适用项目小型项目、简单异步场景中大型生产项目、分布式系统七、新手高频避坑总结1. ❌ 后台任务使用请求内瞬时资源DB会话、请求对象导致报错2. ❌ 依赖 BackgroundTasks 执行长耗时核心业务任务丢失造成数据异常3. ❌ WebSocket 不做异常捕获客户端断开导致服务报错4. ❌ 静态文件目录不存在导致服务启动失败5. ✅ 轻量异步用内置后台任务核心异步任务务必上 Celery6. ✅ WebSocket 必须统一管理在线连接规范处理断开异常7. ✅ 流式响应优先用于大文件、实时输出场景减少内存占用。