你是不是觉得 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 功能客户端发什么我就原样返回什么。/* by 01022.hk - online tools website : 01022.hk/zh/editor.html */ 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(客户端断开连接) except Exception as e: print(f发生异常: {e}) # 记得主动关闭连接 await websocket.close()看着是不是很简单✨ 但这里已经藏着一个坑如果客户端异常断开比如网络闪断receive_text()会抛出WebSocketDisconnect你得捕获它否则程序会崩溃。我曾经没写这个 try结果 uvicorn 进程直接挂掉教训惨痛。 3. 五个让你凌晨三点爬起来填的坑附解决方案 3.1 连接自动断开因为没有心跳很多云服务商比如阿里云、AWS的负载均衡器如果一段时间内没有数据传输会认为连接空闲而把它掐掉。这个时间通常是 60 秒左右。解决方案服务端和客户端都要有心跳机制。最简单的做法是客户端每隔 30 秒发一个 ping 帧或者自定义心跳消息服务端收到后回复 pong。当初我偷懒没做心跳用户画图时停顿超过 1 分钟就掉线被产品经理追着打。 后来加了个定时器世界安静了。/* by 01022.hk - online tools website : 01022.hk/zh/editor.html */ # 服务端处理心跳的伪代码 async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: try: # 设置接收超时如果一段时间没收到任何消息主动发心跳探测 data await asyncio.wait_for(websocket.receive_text(), timeout30) # 处理正常消息 except asyncio.TimeoutError: # 30 秒没收到消息主动发送 ping 探测 await websocket.send_text(__ping__) # 自定义心跳 # 如果客户端没响应会在下一次循环触发 WebSocketDisconnect 3.2 WebSocket 握手时如何携带 TokenWebSocket 的握手是 HTTP 请求所以可以在 URL 参数或者 Header 里带 token。但千万不要在路径里明文传 token会记在日志里推荐的做法用Sec-WebSocket-Protocol或者 Header 里的Authorization。FastAPI 的依赖注入也支持 WebSocket你可以写一个依赖来校验。from fastapi import WebSocket, WebSocketException, status async def get_cookie_or_token(websocket: WebSocket): # 从 query 参数或 header 取 token示例从 query 取 token websocket.query_params.get(token) if token ! secret: await websocket.close(codestatus.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()并捕获异常给客户端返回错误码而不是断开连接。 4. 程序媛的碎碎念从能用→好用→稳如泰山WebSocket 的上手门槛确实低但想在生产环境跑得稳你需要考虑这些✅ 监控每一条连接的存活时间、收发消息数量最好都能通过 metrics 暴露出来比如用 Prometheus。✅ 限流单个用户发消息太快怎么办可以用漏桶或令牌桶限制频率防止被恶意攻击。✅ 断线重连客户端一定要有自动重连机制指数退避策略避免同时重连造成服务器压力。✅ 测试用websockets库写脚本模拟数千个连接压测一下你的服务看资源占用和响应延迟。我还记得有一次我忘记设置max_size限制客户端传了个超大的消息直接把内存撑爆了。后来在 accept 之前加上websocket.max_size 1_048_5761MB世界又清净了。老朋友今天的分享就到这里。如果你也在 FastAPI 里踩过 WebSocket 的坑或者有什么独家秘笈欢迎留言区切磋。⭐️ 觉得有用的话点个赞再走呗这样我下回才有动力写出更多FastAPI 避坑指南。当然关注我也可以毕竟你不知道我下次又会因为什么骚操作给你带来新的实战经验