Websocket

多数 HTTP 请求是一次性的:客户端问,服务器答,连接也许就结束了。WebSocket 升级一条连接,让任一侧随时都能发送。在它和轮询之间选择,是在决定更新该被「推」还是被「拉」。 升级买到什么 WebSocket 的前半段,是一次普通 HTTP 请求去请求「升级」;服务器答允之后,原本一问一答的连接就变成一条持久、双向的通道。这次升级真正买来的,是不再非要等客户端问一次才答——只要数据一有变化,服务端就能主动把消息推过去。由此语义也变了:同一条连接上流动的不再是一串相继的往返,而是连续许多条消息,所以聊天、实时看板那类场景,不必每条都重走一遍请求链,只靠一条低延迟管道就能一直保持。 轮询仍是一个正当的基线 轮询依然是一个完全正当的基线,在下决心用 WebSocket 之前,应当先想想它。如果事件难得出现、又允许略晚到达,那么用普通间隔去读一次数据,会更简单、更贴近缓存、也更方便运营,远没有一套握手与长连接那么多环节。每一次推送都要付出消息与握手的代价,还要让负载均衡器和网络中间件容忍长期空闲的「挂起」。真正需要服务端主动推送、又要低延迟时,再选择 WebSocket,而不是为了追新而追新。 让会话保持健康 一条会话一旦长期打开,你就要负责让它保持健康。定时发心跳或 ping 消息,免得空闲连接被中间层的中间件踢掉;掉线后按照「指数退避」重连,并把断掉的会话当作「页面关闭」来善后——清理状态、取消订阅。更重要的是身份验证必须在那次握手时完成,因为升级之后再没有一次新请求能验明正身。经营一条 websocket 干线,意味着重连、心跳、清理与身份这些准备,远在故障之前就该就绪。 坏掉的样子 空闲连接最先被代理杀死。公司代理、CDN 与负载均衡器常常把静默六十秒的连接直接掐断,而两端都还以为会话好好活着。这种失败是静默的:没有 close 帧到达,客户端把下一条消息推进死连接,服务端还留着一个永远不再读的订阅者。没有心跳,就无法区分「安静的对端」与「已死的对端」,服务端只能慢慢堆积幽灵连接,白白占用文件描述符,还替无人消费的广播买单。 扩容会暴露第二种失败。会话状态长在接受握手的那台机器上:订阅集合、鉴权结果、积压队列都在本地。若粘性会话丢失,重连落到另一台机器,客户端必须从头订阅;若它误以为旧订阅还在,就会静默丢消息。身份是同一个陷阱:令牌只在握手那一刻验过一次,升级之后再也不会随消息同行。 还有一种慢性失败是背压失控。推送端只管往连接里写,从不看对端读得多慢,慢消费者会把服务端内存一点点顶满;反过来客户端也可能被每秒数百条行情淹没,界面卡死还误以为是渲染问题。会话是双向的,任何一侧都该有队列上限与丢弃策略:慢订阅者要么降频采样,要么主动断开,绝不能让一条慢连接拖垮整台机器。 恢复可能补上最后一击。服务端重启,或代理一次掐断数千条连接,所有客户端立刻同时重拨,重试洪峰变成重连风暴,正好打在鉴权与订阅链路上——那恰恰是容量最低的时刻。修法故意无聊:心跳间隔短于代理空闲超时,连续两次错过 pong 就关闭并清理,重连必重新鉴权,订阅必须能用一条消息重建,重连一律指数退避加抖动。 快速验证 把下面程序存为 ws_session_probe.py,运行 python3 ws_session_probe.py。它只用标准库,只绑回环地址,不改进程之外的任何东西。回环 http.server 校验原始升级请求,socketpair 模拟心跳期限。 """Probe WebSocket session semantics with stdlib only: handshake + heartbeat.""" import base64 import hashlib import socket import threading from http.server import BaseHTTPRequestHandler, HTTPServer GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" # RFC 6455 section 1.2 example key and its expected accept value. KEY = "dGhlIHNhbXBsZSBub25jZQ==" EXPECTED_ACCEPT = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" def compute_accept(key: str) -> str: return base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode() class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def do_GET(self): key = self.headers.get("Sec-WebSocket-Key") upgrade = (self.headers.get("Upgrade") or "").lower() connection = (self.headers.get("Connection") or "").lower() ok = ( self.path == "/chat" and key is not None and upgrade == "websocket" and "upgrade" in connection ) if not ok: body = b"expected websocket upgrade" self.send_response(400) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return accept = compute_accept(key.strip()) self.send_response(101, "Switching Protocols") self.send_header("Upgrade", "websocket") self.send_header("Connection", "Upgrade") self.send_header("Sec-WebSocket-Accept", accept) self.send_header("Content-Length", "0") self.end_headers() def log_message(self, *args): pass def read_http_response(sock: socket.socket) -> str: data = b"" sock.settimeout(5) while b"\r\n\r\n" not in data: chunk = sock.recv(4096) if not chunk: break data += chunk return data.decode("latin-1") def main() -> None: # 1. The accept derivation itself must match the RFC 6455 test vector. assert compute_accept(KEY) == EXPECTED_ACCEPT, "accept mismatch" print(f"accept vector ok: {EXPECTED_ACCEPT}") # 2. A real loopback handshake: raw request, assert 101 + accept echo. server = HTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: port = server.server_address[1] with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: sock.sendall( ( "GET /chat HTTP/1.1\r\n" "Host: 127.0.0.1\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {KEY}\r\n" "Sec-WebSocket-Version: 13\r\n" "\r\n" ).encode() ) head = read_http_response(sock) assert "101" in head.splitlines()[0], head assert f"Sec-WebSocket-Accept: {EXPECTED_ACCEPT}" in head, head print("handshake ok: 101 + Sec-WebSocket-Accept matches key") # 3. A wrong path is not a session: it must stay plain HTTP. with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: sock.sendall(b"GET /page HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n") head = read_http_response(sock) assert "400" in head.splitlines()[0], head print("non-upgrade ok: plain GET gets 400, no session created") # 4. Heartbeat semantics over a socketpair: a pong in time keeps the # session, silence past the deadline marks a ghost connection. alive, peer = socket.socketpair() with alive, peer: alive.settimeout(2) peer.settimeout(2) alive.sendall(b"ping-1") assert peer.recv(1024) == b"ping-1" peer.sendall(b"pong-1") assert alive.recv(1024) == b"pong-1" print("heartbeat ok: ping/pong round-trip keeps session alive") peer.settimeout(0.2) try: peer.recv(1024) raise AssertionError("expected silence, got data") except socket.timeout: print("ghost detection ok: missed pong past deadline = stale session") finally: server.shutdown() server.server_close() print("PASS: handshake, rejection, heartbeat, and ghost detection hold") if __name__ == "__main__": main() 脚本断言四条会话语义。接受值推导与 RFC 6455 示例向量一致;格式正确的升级返回 101 并回显匹配的 Sec-WebSocket-Accept,而普通 GET 只得 400 且不创建会话。ping 与 pong 往返让会话存活,超过期限的沉默即判为幽灵连接。运行成功会逐行打印向量、握手、心跳与幽灵判定,最后输出 PASS;任一断言失败即抛异常并以非零状态退出,可直接当作回归门禁。以上断言在 macOS、Python 3.14 下运行通过。该夹具只覆盖握手与心跳逻辑,不复现代理空闲超时、TLS 终结或浏览器重连行为,上线前请在真实链路上复测空闲与风暴场景。 ...