当前位置：首页 > news >正文

FastAPI 实战：WebSocket 从入门到上线，使用避坑指南

news 2026/3/26 17:41:41

💬 你是不是觉得 WebSocket 挺简单的？不就建立个连接，然后 send 和 onmessage 吗？
我当初也这么想，直到我做了一个在线协作小工具，上线第一天晚上，服务器直接崩了，用户疯狂吐槽：“怎么画着画着就断了？” 那一晚，我盯着监控面板，才真正明白——WebSocket 的坑，不在握手，而在长连接的战场上。

📌 本文摘要：从 FastAPI 实战出发，分享在生产环境中用 WebSocket 踩过的坑、填坑的方案，以及那些容易忽略的注意事项。读完你不仅能跑通 Demo，还能让连接更稳定，心里更有底。

🚀 FastAPI 实战：WebSocket 使用避坑指南，让你的长连接稳如老狗

🎯 这篇你会得到什么：

🔹 第一部分：WebSocket 的本质（用一个餐厅的故事讲透）

🔹 第二部分：FastAPI 集成 WebSocket 的最小实战（带代码）

🔹 第三部分：5 个最常见翻车点及解决方案（心跳、认证、并发、部署、异常）

🔹 第四部分：老程序员的碎碎念（经验总结 + 互动）

🍳 1. 先从 WebSocket 是什么聊起（一个餐厅的比喻）

传统的 HTTP 请求，就像你去餐厅点菜：你喊一声“服务员，来份宫保鸡丁”，然后服务员跑去后厨，把菜端给你，一次交易结束。你要再点个米饭，又得喊一次。

而 WebSocket 呢？它就像你直接在餐厅包了个雅间，服务员就站在你桌旁，随时听你吩咐：“加点水”、“拿头蒜”、“结账” —— 服务员一直在线，随时响应，省去了反复呼叫的过程。

FastAPI 对 WebSocket 的支持非常 Pythonic，用起来很顺手，但正因为顺手，容易忽略背后那些“服务员也得休息”、“包间太多会拥挤”的现实问题。

⚡ 2. 最小实战：一个简单的聊天 echo 服务

先来个最基础的，看看 FastAPI 里 WebSocket 长啥样。下面的代码实现了一个 echo 功能：客户端发什么，我就原样返回什么。

from fastapi import FastAPI, WebSocket, WebSocketDisconnectapp = 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("客户端断开连接")except Exception as e:print(f"发生异常: {e}")# 记得主动关闭连接await websocket.close()

看着是不是很简单？✨ 但这里已经藏着一个坑：如果客户端异常断开（比如网络闪断），receive_text() 会抛出 WebSocketDisconnect，你得捕获它，否则程序会崩溃。我曾经没写这个 try，结果 uvicorn 进程直接挂掉，教训惨痛。

💣 3. 五个让你凌晨三点爬起来填的坑（附解决方案）

👉 3.1 连接自动断开？因为没有心跳

很多云服务商（比如阿里云、AWS）的负载均衡器，如果一段时间内没有数据传输，会认为连接空闲而把它掐掉。这个时间通常是 60 秒左右。

解决方案： 服务端和客户端都要有心跳机制。最简单的做法是客户端每隔 30 秒发一个 ping 帧（或者自定义心跳消息），服务端收到后回复 pong。

当初我偷懒没做心跳，用户画图时停顿超过 1 分钟就掉线，被产品经理追着打。😭 后来加了个定时器，世界安静了。

# 服务端处理心跳的伪代码
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:try:# 设置接收超时，如果一段时间没收到任何消息，主动发心跳探测data = await asyncio.wait_for(websocket.receive_text(), timeout=30)# 处理正常消息except asyncio.TimeoutError:# 30 秒没收到消息，主动发送 ping 探测await websocket.send_text("__ping__")  # 自定义心跳# 如果客户端没响应，会在下一次循环触发 WebSocketDisconnect

👉 3.2 WebSocket 握手时如何携带 Token？

WebSocket 的握手是 HTTP 请求，所以可以在 URL 参数或者 Header 里带 token。但千万不要在路径里明文传 token，会记在日志里！

推荐的做法：用 Sec-WebSocket-Protocol 或者 Header 里的 Authorization。FastAPI 的依赖注入也支持 WebSocket，你可以写一个依赖来校验。

from fastapi import WebSocket, WebSocketException, statusasync def get_cookie_or_token(websocket: WebSocket):# 从 query 参数或 header 取 token（示例从 query 取）token = websocket.query_params.get("token")if token != "secret":await websocket.close(code=status.WS_1008_POLICY_VIOLATION)raise WebSocketException("认证失败")return token@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_cookie_or_token)):await websocket.accept()# ...

👉 3.3 连接数一高，服务就卡死？

每个 WebSocket 连接都会占用一个系统文件描述符和内存。默认的 uvicorn 单进程能支撑的连接数有限（取决于你机器配置）。如果直接裸跑，内存可能会被慢慢吃光。

解决方案：

🔹 用 gunicorn + uvicorn workers 启动多进程，注意设置适当的 worker 数量，一般 = CPU 核心数 * 2。

🔹 如果广播消息频繁，不要用简单的全局循环给所有连接发，可以用 asyncio.gather 并发发送，但注意控制并发数，防止突发流量把 CPU 打满。

🔹 考虑使用 Redis 等中间件做消息分发，特别是多进程模式下，一个进程不知道另一个进程管理的连接，需要借助外部广播。

⚠️ 线上教训：我曾用单进程跑了 5000 个连接，结果内存占用 2GB，频繁 GC，最后用 --workers 4 解决了，但广播又成了新问题，后来引入了 Redis pub/sub。

👉 3.4 服务重启时，用户瞬间掉线怎么办？

当你部署新版本，需要重启服务，所有 WebSocket 连接会被粗暴关闭。用户会看到“连接已断开”。这很不优雅。

解决方案： 利用 uvicorn 的 lifespan 事件，在关闭前主动通知客户端（比如发一条“服务即将维护”的消息），并等待几秒再关闭。也可以配合负载均衡的 draining 机制。

# 在 shutdown 事件里做清理
@app.on_event("shutdown")
async def shutdown_cleanup():# 遍历所有活跃连接，发下线通知for connection in active_connections:try:await connection.send_text("server going down, reconnect later")except:pass# 等 1 秒让消息发出去await asyncio.sleep(1)

👉 3.5 文本还是二进制？JSON 还是自定义？

WebSocket 支持文本和二进制帧。如果你们前后端约定用 JSON，记得处理解析异常。我曾经因为前端发了个非法的 JSON，服务端没捕获 json.JSONDecodeError，直接导致连接崩溃。

稳健的做法： 统一用 receive_json() 并捕获异常，给客户端返回错误码，而不是断开连接。