BackgroundTasks 在 FastAPI 的 WebSocket 路由中完全无效，因其依赖 HTTP 请求生命周期和响应返回后的钩子，而 WebSocket 无响应返回阶段，任务无法注册或触发；必须改用 asyncio.create_task() 并注意异常捕获、避免直接操作 websocket、通过队列中转消息。
BackgroundTasks 在 FastAPI 的 WebSocket 路由里完全无效，不是配置问题，是设计限制——它只活在 HTTP 请求生命周期内，而 WebSocket 连接一旦建立，就脱离了那个上下文。
为什么 BackgroundTasks 在 @app.websocket 里不执行
FastAPI 的 BackgroundTasks 机制依赖 Request 对象和响应返回后的清理钩子。WebSocket 没有“响应返回”这一步：连接持续存在，BackgroundTasks 根本没机会注册或触发。你调用 background_tasks.add_task(...) 后函数静默消失，不是报错，是压根没被调度。
asyncio.create_task() 是唯一可靠方案
必须手动把协程扔进事件循环，才能让后台逻辑和 WebSocket 主循环并行跑。但要注意三点：
- 任务需自行捕获异常，否则崩溃无声，连接可能卡死
- 避免在任务里直接操作
websocket实例——它不是线程安全的，且可能在任务运行中途被关闭 - 若需向客户端发消息，建议通过共享队列（如
asyncio.Queue）中转，主 WebSocket 循环统一消费
示例关键片段：
立即学习“Python免费学习笔记（深入）”；
import asyncio
from fastapi import WebSocket
<p>@app.websocket("/ws/{user_id}")
async def websocket_endpoint(websocket: WebSocket, user_id: str):
await websocket.accept()</p><h1>启动后台监听 Redis Pub/Sub</h1><pre class="brush:php;toolbar:false;">task = asyncio.create_task(listen_redis_channel(user_id))
try:
while True:
data = await websocket.receive_text()
# 处理客户端消息
except Exception:
pass
finally:
task.cancel() # 连接断开时清理
try:
await task
except asyncio.CancelledError:
pass
路径参数、查询参数和认证怎么传
WebSocket 路由支持路径参数（如 /ws/{room_id}）和查询参数（如 ?token=abc123），用法和普通路由一致，但注意：
-
Query默认不校验必填，要强制传 token 就得写token: str = Query(...) - 认证失败不能返回 HTTP 状态码，只能抛
WebSocketException(status.WS_1008_POLICY_VIOLATION) - 路径参数类型转换失败会直接断连，无友好提示，建议在
accept()前加try/except
连接管理容易漏掉的清理点
实际部署时最常出问题的是连接泄漏：客户端断开后，websocket 实例没从内存列表里移除，导致后续广播卡顿甚至 OOM。
- 不要依赖
except WebSocketDisconnect:——网络抖动、代理超时等场景下它不一定触发 - 务必在
finally块里做清理，哪怕只是connected_clients.discard(websocket) - 如果用了全局连接池（比如
set[WebSocket]），记得用id(websocket)或自定义 ID 做键，避免对象销毁后残留引用
真正稳定的 WebSocket 服务，90% 的坑不在协议理解，而在连接生命周期的边界处理是否足够“脏”。
