python

《Python编程实战》10.1 WebSocket 与消息协议

从轮询到双向长连接：用标准库真跑一遍 WebSocket 握手与帧格式，掌握 FastAPI 的 @app.websocket 与 WebSocketDisconnect，并设计带类型字段与版本协商的 JSON 消息信封，最后讲清文本 JSON 与二进制的取舍。

Leeting Yan 7 分钟阅读 3380 字

本节目标：搞清 WebSocket 与普通 HTTP 的本质差别，用真实字节看懂握手与帧格式，学会 FastAPI 的 @app.websocket 用法，并设计一套带类型字段与版本协商的消息协议。
适用版本：Python 3.12+（实测 3.14.6）；fastapi 0.143.0、starlette 1.7.0、uvicorn 0.54.0

10.1 WebSocket 与消息协议

第 9 章我们把「谁能进来、能干什么」划清楚了。但前面所有接口都是请求-响应模型：客户端问一句，服务端答一句，然后连接就闲置。聊天、协同编辑、实时行情、进度推送这类场景里，服务端需要主动把消息推给客户端——这正是本章要解决的。

10.1.1 为什么需要 WebSocket

在 WebSocket 之前，服务端要「推」消息只能靠三种笨办法：

方案做法代价
短轮询客户端每秒问一次「有新消息吗」大量空请求，延迟高
长轮询请求挂起直到有数据才返回连接频繁重建，服务端扛不住
HTTP/2 Server Push服务端推静态资源语义是推资源，不是消息通道，已被浏览器弃用

它们的共同问题是每次都要重新走一遍 HTTP 请求头（几百字节到几 KB），而且永远只能客户端先开口。WebSocket 用一个「HTTP 握手 + 之后裸帧」的协议解决这两点：握手复用一次 HTTP 请求完成协议切换（Upgrade），之后这条 TCP 连接上跑的是轻量二进制帧，且双方对等，谁都能随时发。

10.1.2 握手：一次 HTTP Upgrade

WebSocket 连接从一次看起来像普通 GET 的 HTTP 请求开始，关键在于三个头：Upgrade: websocket、Connection: Upgrade、以及一个随机的 Sec-WebSocket-Key。服务端校验后回 101 Switching Protocols。

本机未安装 websockets / wsproto，uvicorn 无法完成真实升级（请求会拿到 404），所以下面用标准库 asyncio 实现一个最小 RFC 6455 服务端，配合裸 socket 客户端抓真实字节：

import asyncio, base64, hashlib, struct

GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

def accept_key(client_key: str) -> str:
    # Sec-WebSocket-Accept = base64(SHA1(client_key + GUID))
    return base64.b64encode(hashlib.sha1((client_key + GUID).encode()).digest()).decode()

async def handle(reader, writer):
    head = await reader.readuntil(b"\r\n\r\n")
    headers = {}
    for line in head.decode().split("\r\n")[1:]:
        if ":" in line:
            k, v = line.split(":", 1)
            headers[k.strip().lower()] = v.strip()
    key = headers["sec-websocket-key"]
    resp = (
        "HTTP/1.1 101 Switching Protocols\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        f"Sec-WebSocket-Accept: {accept_key(key)}\r\n\r\n"
    )
    writer.write(resp.encode())
    await writer.drain()

真跑（客户端发 Sec-WebSocket-Key: AlXrvIYqNGJFDVKkHbvBzQ==，实测输出）：

=== 服务端收到握手请求 ===
  GET /ws HTTP/1.1
  upgrade: websocket
  connection: Upgrade
  sec-websocket-key: AlXrvIYqNGJFDVKkHbvBzQ==
  sec-websocket-version: 13
=== 服务端返回 101 ===
  HTTP/1.1 101 Switching Protocols | Upgrade: websocket | Connection: Upgrade | Sec-WebSocket-Accept: MllJ1zpudP4a4sIxMHI9jpmkGdA=
=== 客户端校验 ===
  Accept 头: MllJ1zpudP4a4sIxMHI9jpmkGdA=
  本地计算: MllJ1zpudP4a4sIxMHI9jpmkGdA=

Sec-WebSocket-Accept 的计算是握手防伪的核心：客户端发一个随机 key，服务端把 key + GUID 做 SHA1 再 base64 回传；客户端本地算一遍比对，对得上才证明对面真的懂 WebSocket，而不是一个把请求当普通 GET 处理的缓存或代理（那种情况会返回 404 或缓存的 HTML）。GUID 是协议写死的常量，防的是中间设备「瞎应答」。

10.1.3 帧协议要点

握手完成后，通信单元变成帧（frame）。一帧的头部只有 2 到 14 字节，几个关键位：

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-------+-+-------------+-------------------------------+
|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
|N|V|V|V|       |S|             |                               |
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +

实测一帧往返（客户端发 JSON，服务端回 ack）：

=== 服务端收到数据帧 opcode=0x1 载荷={"v": 1, "type": "chat", "payload": {"text": "hi"}}
=== 客户端收到响应帧 ===
  第一字节 0x81 FIN=1 opcode=0x1
  掩码位 MASK=0（服务器->客户端不掩码）
  载荷: {"v": 1, "type": "ack", "payload": {"ok": true}}

四个要点，记住了就不会踩坑：

  • opcode 决定帧类型：0x1 文本、0x2 二进制、0x8 关闭、0x9 ping、0xA pong。应用数据只有前两种。
  • FIN 表示分片结束：大消息会被拆成多个帧（第一帧 FIN=0，末帧 FIN=1），接收方要按 opcode 拼接。
  • 掩码方向是单向的：客户端到服务端必须掩码（MASK=1），服务端到客户端禁止掩码。这是为了防代理缓存投毒，不是加密——掩码 key 就在帧里，谁都能还原。
  • 关闭要握手：一端发 0x8（可带 2 字节状态码，实测 1000 表示正常关闭），另一端回 0x8，然后才关 TCP。

10.1.4 FastAPI 的 @app.websocket

框架把这套帧处理全包了，你只面对「收消息 / 发消息」。FastAPI 用装饰器声明 WebSocket 端点：

import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws/echo")
async def echo(websocket: WebSocket) -> None:
    await websocket.accept()                 # 对应 101 响应，必须显式调用
    try:
        while True:
            msg = json.loads(await websocket.receive_text())
            reply = {"v": 1, "type": "echo", "id": msg.get("id"),
                     "payload": {"text": msg["payload"]["text"]}}
            await websocket.send_text(json.dumps(reply, ensure_ascii=False))
    except WebSocketDisconnect as exc:
        print(f"[server] client disconnected code={exc.code}")

用 starlette 的 TestClient.websocket_connect 真跑（实测）：

from starlette.testclient import TestClient

with TestClient(app) as client:
    with client.websocket_connect("/ws/echo") as ws:
        for text in ["你好", "world"]:
            ws.send_text(json.dumps({"v": 1, "type": "chat", "id": 7,
                                     "payload": {"text": text}}))
            print("[client]", ws.receive_text())

[client] {"v": 1, "type": "echo", "id": 7, "payload": {"text": "你好"}}
[client] {"v": 1, "type": "echo", "id": 7, "payload": {"text": "world"}}
[server] client disconnected code=1000

三个必须知道的点：

  1. accept() 不能省。不调用就直接 receive 会报错；它也是你唯一能拒绝连接的地方（比如鉴权失败时 await websocket.close(code=1008)）。
  2. WebSocketDisconnect 是正常路径。客户端关页面就是走这个异常，实测 code=1000。不要把它当错误吞掉——它的 except 块正是你清理连接、通知房间的地方（10.3 的 ConnectionManager.disconnect 就在这里调用）。
  3. receive_text() 是阻塞协程。它 await 直到有消息，不会占 CPU，但同一个 while 里若还要主动推消息，就得用 10.3 讲的任务分工，不能死等。

补充观察：本机 starlette 1.7.0 的 TestClient 会打印 StarletteDeprecationWarning: Using httpx with starlette.testclient is deprecated; install httpx2 instead.——这是测试工具的迁移提示，不影响生产代码。

10.1.5 消息协议设计：JSON 信封

WebSocket 只给「一帧一帧的字节」，它不规定消息长什么样。裸发字符串迟早失控（前端发 {"text":"hi"}、另一个端发 {"msg":"hi"}，服务端到处 if）。工程做法是统一信封（envelope）：

{
  "v": 1,
  "type": "chat",
  "id": 7,
  "payload": { "text": "hi" }
}
字段作用是否必需
v协议版本，用于协商与兼容是
type消息类型，路由到对应 handler是
id请求序号，用于把「回应」对上「请求」可选
payload真正的业务数据是

type 用点号分层（chat、chat.ack、chat.error）比扁平的 CHAT_ACK 更易扩展：前端可以按前缀订阅一类消息。id 解决的是异步回应问题——WebSocket 上没有 HTTP 的「一个请求对一个响应」，多条消息可能乱序回来，靠 id 才能对上号。

10.1.6 类型字段与版本协商

把「解析 + 校验 + 分发」写成一层，所有消息都从这里过。用 match 做类型分发，同时在入口就把版本不对、格式不对的消息挡掉：

import json

SUPPORTED_VERSIONS = {1}
MAX_FRAME_BYTES = 64 * 1024

class ProtocolError(Exception):
    pass

def decode(raw: str) -> dict:
    if len(raw.encode()) > MAX_FRAME_BYTES:          # 防超大帧打爆内存
        raise ProtocolError("frame too large")
    try:
        msg = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ProtocolError(f"malformed json: {exc.msg}") from exc
    if not isinstance(msg, dict):
        raise ProtocolError("envelope must be an object")
    if msg.get("v") not in SUPPORTED_VERSIONS:       # 版本协商
        raise ProtocolError(f"unsupported version: {msg.get('v')!r}")
    if "type" not in msg:
        raise ProtocolError("missing type")
    return msg

def handle(msg: dict) -> dict:
    match msg["type"]:
        case "chat":
            return {"v": 1, "type": "chat.ack", "id": msg.get("id"),
                    "payload": {"ok": True}}
        case "ping":
            return {"v": 1, "type": "pong", "payload": {}}
        case other:
            return {"v": 1, "type": "error",
                    "payload": {"code": "unknown_type", "got": other}}

真跑六种输入（实测）：

  {"v":1,"type":"chat","id":7,"payload":{"text":"hi"}} -> {"v": 1, "type": "chat.ack", "id": 7, "payload": {"ok": true}}
  {"v":1,"type":"ping"}                          -> {"v": 1, "type": "pong", "payload": {}}
  {"v":2,"type":"chat"}                          -> ProtocolError: unsupported version: 2
  {"type":"chat"}                                -> ProtocolError: unsupported version: None
  {"v":1,"type":"nope"}                          -> {"v": 1, "type": "error", "payload": {"code": "unknown_type", "got": "nope"}}
  not json                                       -> ProtocolError: malformed json: Expecting value

注意两种失败的层次不同：v:2 和缺 v 是协议层错误（ProtocolError，应回一条 error 帧甚至直接关闭连接），而 type:nope 是业务层错误（协议合法，只是不认识这个类型，回 error 帧即可）。分清这两层，前端才能决定「是重连还是重发」。

10.1.7 文本 JSON vs 二进制

既然握手和帧都能传任意字节，为什么大多数应用还是选 JSON 文本？对照如下：

维度JSON 文本（opcode 0x1）二进制（opcode 0x2，如 MessagePack/Protobuf）
体积大（字段名重复、数字变字符串）小 30%~70%
可读性浏览器 DevTools 直接看要工具解码
调试成本低高
CPU序列化开销小编解码更省，但需 schema
适用消息量不大、要快速迭代高频行情、大数组、弱网

默认选 JSON，把带宽和延迟问题留到真有瓶颈时再换。要换也不必推倒重来：信封结构不变，只把 payload 的编解码换成 MessagePack 即可，v/type/id 三层逻辑完全复用——这正是先定协议再定编码的价值。

延伸阅读

小结

  • WebSocket 用一次 HTTP Upgrade 握手换来一条双向、低开销的长连接，适合服务端主动推送的场景。
  • Sec-WebSocket-Accept = base64(SHA1(key + GUID)) 是握手防伪的核心，客户端要本地校验。
  • 帧层面记住四件事：opcode 定类型、FIN 定分片、掩码单向、关闭要握手。
  • FastAPI 用 @app.websocket + WebSocketDisconnect，accept() 必须显式调用，断连是正常路径不是错误。
  • 协议要设计统一 JSON 信封：v 管版本、type 管路由、id 管对应、payload 管业务。
  • 分清协议层错误（版本/格式，可关连接）与业务层错误（未知类型，回 error 帧），前端才能正确决定重连还是重发。
  • 默认用 JSON 文本，先定协议再定编码，将来换二进制只动 payload。

本节把「双向通道」和「消息长什么样」讲透了，但还没解决「服务端主动推」的另一条路——SSE。下一节看它如何用一条只读的 HTTP 流实现推送，以及什么时候它比 WebSocket 更合适。

阅读导航：上一节：输入校验、依赖供应链与漏洞加固 · 下一节：SSE 与流式响应 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「python」更多文章

  1. 《Python高级编程》目录
  2. 《Python高级编程》11.3 PEP 流程与版本迁移策略
  3. 《Python高级编程》11.2 嵌入式与自由线程运行时