★ FEATURED ARTICLE 做后端这几年我越来越觉得 WebSocket 是个“人人都听说过、没几个人真正写明白”的东西。身边不少同事都能用 Python 随手跑通一个 WebSocket 的 hello world但真到了上线阶段——连接莫名其妙断开、消息偶尔丢失、单机连接数上不去、进程突然卡死——能讲清楚原因的人就寥寥无几了。这篇就把 Python WebSocket 从原理到实战完整拆开讲它为什么会出现、握手和帧在底层怎么流动、Python 里到底该用哪个库、真实的聊天服务和广播服务怎么写以及那些线上最容易翻车的细节。适合刚学会 Python 异步编程、准备把实时功能搬到线上的开发者也适合那些已经被生产环境教育过、想回头补协议课的人。WebSocket 本质上就是“一条长连接上的全双工消息通道”但真要用好它得先懂 HTTP 的边界在哪、帧格式怎么解析、Python 的事件循环怎么配合网络 I/O。下面按一条完整链路往下走从协议层讲到部署踩坑全程有代码、有命令、有可以“抄作业”的方案。1. 先搞清楚 HTTP 为什么吃不消WebSocket 出现前的痛点1.1 请求-响应模型的死结HTTP 从设计之初就是“请求-响应”模型客户端必须主动发一次请求服务端才能回一次响应。服务端永远不能主动向客户端推数据。这个模型在网页浏览场景下非常合理但一旦遇到实时性需求就开始露馅。比如做一个实时通知系统用户正在页面上等一条新的报价、一条新消息、一段日志输出。如果用 HTTP服务端有数据了也不能推给浏览器只能等下一次请求。客户端就得反复问“有变化吗”服务端每次都回复“暂时没有”。我要做第一个实时推送项目时就是这么干的——前端每两秒发一次 Ajax服务端查一遍数据库没有新数据就返回空。功能是跑通了但业务方反馈延迟大、服务器负载高高峰期数据库查询频繁得让人心慌。这里有个更深的限制HTTP/1.1 虽然支持 keep-alive 长连接但一条连接同一时刻只能串行处理请求而且消息格式是“头体”的文本结构每次通信都有几百字节的冗余头。高频轮询下大量资源浪费在传输“我知道了吗”这句话本身。1.2 轮询与长轮询能用但谁用谁知道轮询Polling就是固定间隔重复发请求。它的问题很直观主动拉取会有延迟窗口比如每 2 秒拉一次数据在第 0.5 秒产生用户就得等到第 2 秒才看到。想减少延迟就只能缩短间隔缩短间隔服务器就扛不住。为了克服这个矛盾有人想到了长轮询Long Polling。长轮询的思路是客户端发一个请求服务端不立即返回而是把请求“挂住”等有新数据了再响应客户端收到响应后立刻发下一个请求。这样能实现接近实时的推送代价是每个请求都占着一个持久连接服务端需要为每一个“挂住的请求”保存上下文。当一个系统里有大量长轮询客户端时连接数的压力、超时重连的复杂度都很高。而且长轮询本质上仍是“半双工”服务端想连续发两条消息客户端就得连续发起两个请求消息密集时容易乱套。我后来在一个物联网设备状态同步项目里用过长轮询虽然能出效果但每次新设备接入都要多维护一个 session代码复杂度直线上升。那时我就明白实时通信不是靠“HTTP 技巧”能优雅解决的通信模型本身需要换。1.3 WebSocket 不是魔法它是对 TCP 的合理利用WebSocket 解决痛点的思路很直接既然 TCP 本身就支持双向、同时、持续的数据流动那就在应用层定义一个“长连接消息协议”让客户端和服务端在建立一条 TCP 连接后可以任意时刻互发消息。注意WebSocket 和 HTTP 是并列的应用层协议不是 HTTP 的替代品。它只是在建立连接时“借用”了 HTTP 的握手流程让连接能穿透已有的 Web 基础设施比如网关、负载均衡。一旦握手完成HTTP 的角色就结束了后续的数据交换按照 WebSocket 自己的帧格式进行是真正的全双工。所以WebSocket 非常适合这些场景聊天、实时协作编辑、行情推送、多人游戏、服务端任务进度、日志实时流。它的核心价值是两端地位对等谁有数据谁就发不再需要请求-响应这种刻板的回合制对话。2. 协议层拆解握手与数据帧到底怎么流动2.1 一次升级握手的完整流程WebSocket 连接的建立本质上是一次 HTTP Upgrade 请求。聊天里前端初始化一个 WebSocket底层其实发出来的是这样一个请求头GET /chat HTTP/1.1 Host: example.com Upgrade: websocket Connection: Upgrade Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ Sec-WebSocket-Version: 13这里最关键的字段是Sec-WebSocket-Key它是客户端随机生成的 16 字节、经过 Base64 编码的字符串用途是让服务端证明“我真的支持 WebSocket”而不是随便什么服务端都能升级。服务端收到后会把Key拼上一个固定 GUID258EAFA5-E914-47DA-95CA-C5AB0DC85B11然后做 SHA-1 哈希再 Base64 编码得到Sec-WebSocket-Accept返回给客户端HTTP/1.1 101 Switching Protocols Upgrade: websocket Connection: Upgrade Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbKxOo看到101这个状态码说明协议切换成功底层那条 TCP 连接从此刻开始不再按 HTTP 解析而是按 WebSocket 帧解析。整个握手过程很短但有个细节经常被忽略握手阶段必须是一次完整的 HTTP 请求网关、负载均衡必须能正确识别Upgrade头。很多 “WebSocket 连接失败” 的线上事故其实卡在这一步的代理配置上这一点我后面专门讲。2.2 帧格式一个字节一个字节地读握手完成后数据不再有“头体”的文本分隔而是二进制帧格式。理解帧格式才能真正理解为什么 WebSocket 能适配各种消息类型也方便你排查抓包时看到的原始字节。每个帧从两个字节开头。第一个字节的高 1 位是 FIN表示这是不是消息的最后一帧紧接着 3 位是 RSV1-RSV3一般置 0低 4 位是 opcode标明帧类型。第二个字节的高 1 位是 MASK表示 payload 是否做了掩码低 7 位是 payload 长度如果这个值等于 126那么长度用后面 2 个字节表示如果等于 127用后面 8 个字节表示。字段位置含义FIN首字节最高位1 表示消息边界结束0 表示还有后续分片opcode首字节低 4 位0x1 文本、0x2 二进制、0x8 关闭、0x9 Ping、0xA Pong、0x0 续帧MASK第二字节最高位客户端发服务端必须为 1服务端发客户端必须为 0payload len第二字节低 7 位126 时后续 2 字节是长度127 时后续 8 字节是长度masking key4 字节在 payload 之前用于逐字节异或还原数据这里有个经常被误解的点为什么客户端给服务端的数据必须加掩码而服务端给客户端的不加协议设计的官方解释是为了防止缓存污染攻击。早期一些 HTTP 代理会把 WebSocket 流量误当作普通 HTTP 响应缓存攻击者可以通过构造特定字节序列让代理把伪造的内容当作缓存结果返回给其他用户。客户端加掩码后字节被打乱就算被缓存也不具备直接欺骗其他用户的能力。虽然现代代理基本都正确处理 WebSocket 了但这个规则一直保留着自己实现协议时必须遵守。我建议你如果对这块感兴趣可以抓包看一眼一条文本消息的实际数据是81 8b 4a 3c 2d 1e 6a 0b 0d 4a 2c 5e 0f 1d这样的一串其中81就是 FIN1、opcode0x18b是 MASK1、长度 11 字节。后面 4 个字节是掩码再往后才是异或还原后的真实文本。2.3 控制帧的正确用法ping、pong、closeWebSocket 的控制帧有三个比较常用Ping0x9、Pong0xA、Close0x8。Ping/Pong 的作用是保活和探测。网络连接看起来还在但物理链路可能早就断了中间路由器或网关也可能默默释放空闲连接。常规做法是服务端定期发 Ping客户端只要能回一个 Pong就说明链路通着。Close 帧可以携带状态码和原因文本比如1000表示正常关闭1001表示服务端要走了1008表示策略违规、消息不合规。收到 Close 帧后一般应该回一个 Close 帧然后关闭 TCP。注意完整的关闭流程是四次挥手级别的协商不要只发一个 Close 就立刻断开底层连接否则对端读到的是异常退出前端会弹“连接被意外断开”体验很差。3. Python 生态选型websockets、FastAPI、aiohttp 到底该用哪个3.1 三套方案的优缺点对比Python 标准库里没有 WebSocket 实现第三方库选择倒是不少。我实际用下来主要值得考虑的是这三个方向方案特点适合场景websockets 库专精 WebSocket基于 asyncioAPI 简洁、对协议控制细自研服务端、写协议层工具、轻量推送服务FastAPI Starlette自带 WebSocket 路由和 HTTP 接口共用一套服务已有 FastAPI 的项目、想用同一框架同时管 REST 和 WSaiohttp老牌异步框架也有 WebSocket 支持已经在用 aiohttp 的项目、需要自定义底层行为websockets 库的优点是“小而美”。它不做业务层抽象就是一个异步 WebSocket 客户端/服务端连接、收发、心跳都由你掌控。缺点是你要自己管理连接集合、自己搞广播、自己处理重连框架帮不上忙。FastAPI 则把 WebSocket 当成了框架的一等公民依赖注入、路由、自动文档都能和 HTTP 共用代码结构更整洁。它底层依赖 Starlette 的 WebSocket 实现性能和稳定性都不错适合在同一个项目里同时提供 REST API 和实时通道。aiohttp 我也用过一阵它的 WebSocket API 比 websockets 库历史更早但有些细节比如关闭握手、到websocket.close()的时机写得不够直观除非你本来就在用 aiohttp否则我不会为了 WebSocket 单独引入它。3.2 为什么异步是 WebSocket 的天然底座WebSocket 服务的特点是“连接多、消息稀”。一个聊天服务可能有数万条在线连接但每条连接的活跃频率很低大部分时间都在空等消息。用传统线程模型处理这种连接每来一个连接就开一个线程1 万个连接就是 1 万个线程——光是线程栈占用的内存就是几个 GB更不用说线程切换带来的上下文开销。用进程模型就更不现实了。Python 的 asyncio 解决这个问题的方式是单线程加事件循环。所有连接都在一个事件循环里注册协程执行到网络 I/O 时主动让出控制权等这个连接有数据可读时再继续。你可以想象一个餐厅场景线程模型是给每个顾客配一个服务员寸步不离地站着等异步模型是一个服务员同时记下几十桌顾客的需求哪个桌按铃了他才过去。连接越多异步模型的优势越明显。这也是为什么 Python 做 WebSocket 服务时几乎绕不开“异步”二字。同步的requests、time.sleep、数据库同步驱动在事件循环里都是灾难我后面踩坑清单里会详细讲。3.3 选型建议我的建议很简单如果你要用 Python 从零写一个 WebSocket 服务先学 websockets 库因为它最贴近协议能帮你建立正确的底层直觉如果你的项目已经是 FastAPI直接在里面加 WebSocket 路由即可不需要再单独开一个服务如果你想做的是跨语言、包含房间和事件广播的完整实时平台那么不要直接手写去看看成熟的实时消息框架会更省力——但前提是你已经理解本文讲的这些底层原理否则出了问题根本不知道从哪里排查。选型不是越重越好而是和你项目的其他部分能顺畅整合。一直在折腾基础设施的人在这方面往往吃过亏才明白选一个“看起来功能全”的东西远不如选一个“自己完全能控制”的东西。4. 实战一用 websockets 库写一个能跑的广播服务4.1 先跑通最小可用的服务端和客户端先用 websockets 库写一个最小服务端。假设你已安装pip install websockets服务端代码import asyncio import websockets CONNECTED set() async def handler(websocket): # websockets 10.0 的 handler 只接收一个参数 CONNECTED.add(websocket) try: async for message in websocket: print(f收到: {message}) # 广播给所有已连接客户端 await broadcast(fecho: {message}) except websockets.ConnectionClosed: print(客户端断开) finally: CONNECTED.discard(websocket) async def broadcast(message: str): for ws in list(CONNECTED): try: await ws.send(message) except websockets.ConnectionClosed: CONNECTED.discard(ws) async def main(): async with websockets.serve(handler, 127.0.0.1, 8765): await asyncio.Future() # 一直运行 if __name__ __main__: asyncio.run(main())客户端import asyncio import websockets async def main(): async with websockets.connect(ws://127.0.0.1:8765) as ws: await ws.send(hello server) while True: message await ws.recv() print(message) asyncio.run(main())先跑起来你会看到服务端把消息回显给所有连接。这套代码虽然简单但已经包含了 WebSocket 服务端最重要的三件事接收连接、管理连接、发送消息。注意await websocket.send()和await websocket.recv()前面必须有await漏掉它时消息可能还没发出去协程就结束了。4.2 连接管理不要 set 进去就忘了清理上面的代码里我用一个全局set保存所有连接。广播时为什么用list(CONNECTED)而不是直接for ws in CONNECTED因为广播过程中如果有客户端断开async for message的循环会触发ConnectionClosed然后finally里执行CONNECTED.discard(ws)这相当于在遍历集合的同时修改集合会抛RuntimeError: Set changed size during iteration。用list()复制一个快照再遍历是最稳妥的做法。连接对象放进集合后必须在连接结束时移除。最容易忽略的是网络异常不会让协程立刻抛出异常必须等到下一次收发时才发现。所以除了try...finally最好还要有一个心跳任务定期检查连接活性把那些已经“僵死”的连接从集合里踢出去否则内存和文件描述符都会慢慢泄漏。我见过一个真实的服务连接集合里堆积了上千个早已断开的连接对象每次广播都要遍历一遍这些“僵尸”越跑越慢最后卡死。清理连接这件事看起来只是discard一行但如果少了这行线上就是事故。4.3 心跳保活网络断开不是立刻能感知的TCP 连接断开的感知存在延迟。拔掉网线、手机切飞行模式、中间路由器重启这些情况客户端和服务端不会立刻收到 TCP 的 RST 包连接对象会一直挂在集合里。解决办法就是应用层心跳。最常用的模式是服务端定期 Ping如果一段时间内没有收到对应 Pong就认为连接死了。websockets 库提供了websocket.ping()方法不需要你自己构造帧async def heartbeat(ws, interval20, timeout10): while True: await asyncio.sleep(interval) try: await asyncio.wait_for(ws.ping(), timeouttimeout) except (asyncio.TimeoutError, websockets.ConnectionClosed): await ws.close() break启动时给每个连接asyncio.create_task(heartbeat(websocket))这样就是独立的协程不阻塞主收发循环。另一个模式是服务端不主动 Ping而是用asyncio.wait_for(websocket.recv(), timeout60)包裹接收逻辑60 秒内没收到任何消息就超时断开。后者实现起来更简单但消息比较频繁的服务其实不需要额外心跳“某段时间内没消息就检查一下”已经足够。5. 实战二用 FastAPI 实现带房间的聊天服务5.1 消息协议先约定 JSON别让前端猜广播服务只能 “无脑复制消息”真实业务里还需要区分消息类型、消息归属哪个房间、发送者是谁。我的习惯是通信格式统一用 JSON并且顶层结构固定{ type: join, room: python, user: alice, text: 大家好 }type字段标记动作join加入房间、msg发言、leave离开、ping/pong做应用层心跳。字段名和类型要写清楚前后端按同一份协议联调。千万不要图省事直接把数据库表结构序列化后丢给前端后期扩展时你会被消息格式不一致折磨疯。协议还有个好处是便于限流和审计入口统一解析 JSON不合法的直接回一个错误类型消息再顺手掐掉连接。如果不做这层封装客户端随便发什么进来你的业务代码都要解析、防御安全性和可维护性会一起崩。5.2 RoomManager把连接按房间组织起来用 FastAPI 写带房间的聊天核心是一个连接管理器。它维护一个字典房间名 - 该房间的连接集合。加入房间、离开房间、向房间广播都是对这个字典的操作。直接看代码from fastapi import FastAPI, WebSocket, WebSocketDisconnect import json app FastAPI() class RoomManager: def __init__(self): self.rooms {} async def connect(self, room: str, ws: WebSocket): await ws.accept() self.rooms.setdefault(room, set()).add(ws) def disconnect(self, room: str, ws: WebSocket): if room not in self.rooms: return self.rooms[room].discard(ws) if not self.rooms[room]: del self.rooms[room] async def broadcast(self, room: str, data: dict): if room not in self.rooms: return # 复制一份快照避免遍历时集合被修改 for ws in list(self.rooms[room]): try: await ws.send_json(data) except Exception: self.disconnect(room, ws) manager RoomManager() app.websocket(/ws/chat/{room}) async def chat_endpoint(ws: WebSocket, room: str): await manager.connect(room, ws) join_msg {type: system, text: f用户已加入房间 {room}} await manager.broadcast(room, join_msg) try: while True: data await ws.receive_json() data[room] room await manager.broadcast(room, data) except WebSocketDisconnect: manager.disconnect(room, ws) leave_msg {type: system, text: 用户已离开房间} await manager.broadcast(room, leave_msg)这段代码里有一个关键点await ws.receive_json()在客户端断开时抛出的通常是WebSocketDisconnect异常由 FastAPI 捕获所以disconnect在异常处理里调用。不要把连接的清理放在finally里再做一次否则广播离开消息时会发现连接已经不在集合里。如果消息量很大broadcast里逐个await ws.send_json()会导致消息在房间内不同连接之间出现先后顺序差。对大多数聊天场景这不是问题但如果业务要求同一房间内所有客户端收到的消息顺序完全一致可以把广播改成“每房间一个发送队列 一个后台任务”的模式所有消息先放入队列由单独任务串行发送。这样可以保证同一房间的顺序代价是代码复杂度上了一个台阶。5.3 浏览器端的接入细节前端接入比你想的简单不少。浏览器自带 WebSocket API不需要额外库const ws new WebSocket(ws://${location.host}/ws/chat/python); ws.onopen () { ws.send(JSON.stringify({type: join, user: alice, text: hello})); }; ws.onmessage (event) { const data JSON.parse(event.data); // 渲染到页面 }; ws.onclose () { // 这里做重连指数退避 };这里有两个细节值得注意第一浏览器的 WebSocket 连接会自动带上 Origin 请求头服务端最好校验一下这个头防止其他域的恶意页面往你的服务发消息。第二浏览器端无法自定义Sec-WebSocket-Key和握手相关头所以所有身份认证信息只能在 URL 参数或握手后的第一条消息里传递不要幻想在前端设置 Authorization 头。断线重连是前端的硬需求。不要用固定 2 秒重连万一是服务端短暂重启或者网络抖动一堆客户端同时重连会造成“羊群效应”。指数退避重连通常是更好的选择第 1 次 1 秒、第 2 次 2 秒、第 3 次 4 秒最多 30 秒封顶。重连成功后前端需要重新发 join 消息把自己加回房间否则虽然连接建立了服务端却不知道这个连接属于哪个房间。6. 踩坑清单线上翻车最多的六个细节6.1 网关层超时和升级头最常见的线上事故就是服务正常、代码没问题但 WebSocket 连接就是建不起来或者几秒就断。排查第一站永远在反向代理层。我见过一个 Nginx 配置HTTP 接口一切正常WebSocket 却完全握手失败原因是proxy_pass后面的配置里没有显式声明 Upgrade 头。标准的 Nginx WebSocket 转发配置本质上要做三件事开启 HTTP/1.1、透传 Upgrade、透传 Connection。一个基本配置长这样location /ws/ { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }proxy_read_timeout特别容易踩坑。Nginx 默认 60 秒没有数据就会断开后端连接如果你的应用没有心跳长连接在 60 秒后全都会被 Nginx 杀掉。前端表现就是 1 分钟整WebSocket 准时断开。解决方案要么把 timeout 调大要么在应用层保证有心跳数据流让连接不会长时间“静默”。6.2 多进程部署内存里的连接不在同一个世界另一个搬石头砸自己脚的操作是给 uvicorn 直接开多个 workeruvicorn app:app --workers 2HTTP 请求可以在多进程间负载均衡但 WebSocket 连接建立后是绑定在某个 worker 进程里的。如果 worker 是 2 个进程进程 A 里的连接集合和进程 B 里的集合互不可见。从进程 A 发起的广播进程 B 上挂着的客户端根本收不到。解决思路有两种。第一种WebSocket 服务只开单 worker靠单进程内的协程数量撑高并发——在很多场景下完全够用第二种把“房间到连接”的映射关系搬到所有 worker 共享的外部消息通道里一处广播、多处转发。后者需要额外组件架构会重不少但对于真正需要横向扩展的系统几乎是必经之路。理解这个坑的关键在于WebSocket 是有状态的长连接它不像一次 HTTP 请求那样无状态、可以被任一个后端实例处理。部署模型必须围绕“连接落在哪、消息发到哪”来设计而不是想当然地多开几个进程。6.3 僵尸连接与内存黑洞我在 4.3 里讲过心跳但这里还想再强调一遍。线上最常见的故障链条是客户端网络异常没有正常关闭 TCP服务端连接对象一直留在集合里每次广播推送数据都会触发慢超时、重试、异常堆积内存缓慢上涨最后 OOM。防御手段有三层。第一层是应用层心跳超时主动关闭。第二层是给广播发送设计超时上限不要无限等待。第三层是定期打印连接集合大小和内存占用设置监控告警。大多数团队不是不知道心跳而是“知道但没做”总觉得网络断开会第一时间报异常。实际测试会告诉你一个拔掉网线的客户端TCP 连接可能要几分钟甚至更久才会真正报错这期间它一直被视为“在线用户”。6.4 阻塞调用拖垮整个事件循环把这个坑单独拿出来是因为它对 asyncio 项目来说极其致命。在事件循环里所有协程是协作式调度的谁都不能被占用太久。如果在 WebSocket handler 里写了这样的代码import time import requests async def handler(ws): while True: data await ws.recv() # 错误示例同步请求会卡住整个事件循环 response requests.get(https://third-party-api) await ws.send(response.text)那么当requests.get阻塞时整个事件循环里的所有连接都会一起卡住。表现就是一个客户端发了一条慢请求整个服务器的所有 WebSocket 连接都收不到消息像“雪崩”一样。正确做法是换httpx.AsyncClient或aiohttp远程调用也走异步数据库查询同样要选异步驱动。如果需要调用不支持异步的老库就把任务扔到线程池里用asyncio.to_thread包一层至少不阻塞循环。这几乎是每个 asyncio 服务上线前必查的一项。6.5 忘记 await 和消息大小上限还有一类坑是代码写法带来的。比如await ws.send(msg)写成了ws.send(msg)。在 asyncio 里协程不会被调用直到被 await漏掉 await 的结果是消息根本没发出去只在日志里留下一个 “coroutine was never awaited” 的警告。这个警告很容易被忽略尤其是在循环里不是每条消息都会警告表现特别迷惑。另一个容易忽略的是消息体大小和速率限制。websockets 库默认限制单条消息 1MB超出会断开连接而 FastAPI/Starlette 的 WebSocket 默认对单条消息大小也有约束。如果你的业务要传大图片、大日志一定要显式调大上限并加上服务端速率限制。否则恶意客户端可以疯狂发消息把你的内存和带宽打满。真实线上环境永远要假设有客户端会不按规矩来。7. 连接数、内存与压测评估服务承载能力的实用方法7.1 连接与内存的粗略估算很多人问“我能扛多少连接”这个问题没有标准答案但可以做合理估算。每个 WebSocket 连接在你的 Python 进程里主要开销包括连接对象、收发缓冲区、异步任务栈。粗略来说每个连接的内存占用在 8KB 到 20KB 之间。1 万连接大约消耗 100MB 到 200MB 内存10 万连接就是 1GB 到 2GB 量级。这只是应用层内存还没算操作系统 socket 缓冲区和文件描述符的占用。所以你会发现Python WebSocket 服务通常不是先被 CPU 打死的而是被内存和文件描述符限制住。需要确认操作系统的进程文件描述符上限ulimit -n如果显示 1024那你的服务最多就可能只能保持几百个连接因为每个 socket 都占用一个 fd加上标准输入输出、日志文件1 万连接根本不可能到达。要调大可以临时设置ulimit -n 65535或者按规范方式修改系统限制。7.2 消息积压、背压与限流连接多了之后另一个问题随之而来如果广播速度跟不上生产速度消息会在哪里堆积答案是 socket 缓冲区。你没有直接控制这个缓冲区的接口但它就是会默默变大最终表现为内存上涨、消息延迟严重。更麻烦的是如果某个客户端消费速度极慢服务端向它发送时会阻塞await ws.send()拖慢整个广播循环。应对思路是“背压”。把发送看成一条管道生产消息的一方和消费消息的一方速度不一致时不能让生产方无限发下去。可以在连接对象上维护一个待发送队列并设置上限队列满了就丢弃或者断开该连接到达限流目的# 简化示意每个连接有一个大小为 100 的发送队列 async def safe_send(ws, message, max_queue100): try: await asyncio.wait_for(ws.send(message), timeout2) except (asyncio.TimeoutError, websockets.ConnectionClosed): raise真正的生产级系统一般还有更细的限速按用户、按房间、按全局频率计数配合令牌桶算法。WebSocket 能做到高并发但“高并发”从来不等于“无上限地同时推大量数据”把背压和限流提前设计进去比上线后靠监控救火靠谱得多。7.3 自写压测脚本知道峰值在哪压测 WebSocket 不能用简单的 HTTP 压测工具。常用方式有这么几种用现成的命令行工具 websocat 或wscat --connect ws://...手动模拟交互更大规模的连接压测可以自己写一个 Python 一脚本用asyncio.gather同时发起足够多的连接。下面是一个简单的连接压测框架import asyncio import websockets async def one_client(index, duration): try: async with websockets.connect(ws://127.0.0.1:8765) as ws: await ws.send(fclient {index}) # 等 duration 秒 await asyncio.sleep(duration) except Exception as e: print(fclient {index} failed: {e}) async def main(): tasks [one_client(i, 60) for i in range(5000)] await asyncio.gather(*tasks, return_exceptionsTrue) asyncio.run(main())这种脚本能帮你快速摸清连接数上升到多少时开始出现连接失败、CPU 和内存的拐点在哪、瓶颈是在 Python 进程还是在系统的文件描述符上。注意压测时要站在“预演线上流量”的角度去设计场景不要只测连接建立还要测消息广播、心跳保活、慢客户端对整体的拖累。只有把最坏情况测过一遍上线才有底气。我自己的习惯是先跑小规模连通性测试再按预计峰值的 1.5 倍到 2 倍做压力测试同时盯住三项指标——内存曲线、连接数抖动、消息延迟分位数。如果这三项在持续 10 分钟的压测里保持稳定才敢把服务放到生产环境里。WebSocket 服务不像传统 HTTP 接口那样“请求结束就完事”它是长时间运行、有状态、需要持续观察的所以上线前的压测时长也要足够长不能只跑 30 秒就下结论。
阅读完成 · 觉得有帮助？
