做后端这几年，我越来越觉得 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拼上一个固定 GUID：

258EAFA5-E914-47DA-95CA-C5AB0DC85B11

然后做 SHA-1 哈希，再 Base64 编码，得到Sec-WebSocket-Accept返回给客户端：

HTTP/1.1 101 Switching Protocols Upgrade: websocket Connection: Upgrade Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

看到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 个字节表示。

这里有个经常被误解的点：为什么客户端给服务端的数据必须加掩码，而服务端给客户端的不加？协议设计的官方解释是为了防止缓存污染攻击。早期一些 HTTP 代理会把 WebSocket 流量误当作普通 HTTP 响应缓存，攻击者可以通过构造特定字节序列，让代理把伪造的内容当作缓存结果返回给其他用户。客户端加掩码后，字节被打乱，就算被缓存也不具备直接欺骗其他用户的能力。虽然现代代理基本都正确处理 WebSocket 了，但这个规则一直保留着，自己实现协议时必须遵守。

我建议你如果对这块感兴趣，可以抓包看一眼：一条文本消息的实际数据是81 8b 4a 3c 2d 1e 6a 0b 0d 4a 2c 5e 0f 1d这样的一串，其中81就是 FIN=1、opcode=0x1，8b是 MASK=1、长度 11 字节。后面 4 个字节是掩码，再往后才是异或还原后的真实文本。

2.3 控制帧的正确用法：ping、pong、close

WebSocket 的控制帧有三个比较常用：Ping（0x9）、Pong（0xA）、Close（0x8）。

Ping/Pong 的作用是保活和探测。网络连接看起来还在，但物理链路可能早就断了；中间路由器或网关也可能默默释放空闲连接。常规做法是服务端定期发 Ping，客户端只要能回一个 Pong，就说明链路通着。

Close 帧可以携带状态码和原因文本，比如1000表示正常关闭，1001表示服务端要走了，1008表示策略违规、消息不合规。收到 Close 帧后，一般应该回一个 Close 帧，然后关闭 TCP。注意，完整的关闭流程是四次挥手级别的协商，不要只发一个 Close 就立刻断开底层连接，否则对端读到的是异常退出，前端会弹“连接被意外断开”，体验很差。

3. Python 生态选型：websockets、FastAPI、aiohttp 到底该用哪个

3.1 三套方案的优缺点对比

Python 标准库里没有 WebSocket 实现，第三方库选择倒是不少。我实际用下来，主要值得考虑的是这三个方向：

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(f"echo: {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, interval=20, timeout=10): while True: await asyncio.sleep(interval) try: await asyncio.wait_for(ws.ping(), timeout=timeout) except (asyncio.TimeoutError, websockets.ConnectionClosed): await ws.close() break

启动时给每个连接asyncio.create_task(heartbeat(websocket))，这样就是独立的协程，不阻塞主收发循环。另一个模式是服务端不主动 Ping，而是用asyncio.wait_for(websocket.recv(), timeout=60)包裹接收逻辑：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 直接开多个 worker：

uvicorn app:app --workers 2

HTTP 请求可以在多进程间负载均衡，但 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_queue=100): try: await asyncio.wait_for(ws.send(message), timeout=2) 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(f"client {index}") # 等 duration 秒 await asyncio.sleep(duration) except Exception as e: print(f"client {index} failed: {e}") async def main(): tasks = [one_client(i, 60) for i in range(5000)] await asyncio.gather(*tasks, return_exceptions=True) asyncio.run(main())

这种脚本能帮你快速摸清：连接数上升到多少时开始出现连接失败、CPU 和内存的拐点在哪、瓶颈是在 Python 进程还是在系统的文件描述符上。注意压测时要站在“预演线上流量”的角度去设计场景，不要只测连接建立，还要测消息广播、心跳保活、慢客户端对整体的拖累。只有把最坏情况测过一遍，上线才有底气。

我自己的习惯是：先跑小规模连通性测试，再按预计峰值的 1.5 倍到 2 倍做压力测试，同时盯住三项指标——内存曲线、连接数抖动、消息延迟分位数。如果这三项在持续 10 分钟的压测里保持稳定，才敢把服务放到生产环境里。WebSocket 服务不像传统 HTTP 接口那样“请求结束就完事”，它是长时间运行、有状态、需要持续观察的，所以上线前的压测时长也要足够长，不能只跑 30 秒就下结论。