封面所示及本文讨论的是 HTTP/1.1 升级流程；其他 HTTP 版本的建立方式不同。HTTP 描述请求与响应，并不要求每个请求都新建一条 TCP 连接。WebSocket 提供面向消息的双向会话。选择它而不是轮询，是为了让任一端及时主动发送更新，同时接手会话身份、队列、存活判断与恢复的责任。连接建立只是起点，后续的资源与状态也必须有人管理。
升级买到什么
本文的练习限定在 HTTP/1.1。服务端接受开启握手后返回 101，此后连接中传输的是 WebSocket 帧。检查 Upgrade 与 Connection 时，要匹配列表中完整的成员，并按协议规定忽略大小写；仅仅发现 notupgrade 中含有 upgrade 子串不能算通过。还要检查版本为 13，以及 Base64 密钥解码后恰好十六字节。接受值证明对方理解了握手，却不能证明用户身份。RFC 6455，第 4.2 与 10.5 节。
101 没有 HTTP 消息体，不允许携带 Content-Length，值为零也不例外；同样不能发送 Transfer-Encoding。头部结束后的字节属于 WebSocket 分帧，不能把它理解为普通空响应后随意追加的一段字符串。这种边界必须由实际收发与解析来验证。RFC 9110，第 8.6 节 与 RFC 9112，第 6.1 节。
HTTP/2 的开启流程不同。RFC 8441 在服务端声明支持后，通过一条流上的扩展 CONNECT 建立会话，使用 :protocol websocket 和成功的 2xx 响应，不沿用 HTTP/1.1 的 Upgrade、Connection 或密钥与接受值交换。其他流仍可共享底层连接。下面的练习不验证这一机制，不能拿它的 101 断言去判断 HTTP/2 实现是否正确。RFC 8441，第 3 至 5 节。
轮询仍是一个正当的基线
当更新不频繁、允许一定延迟时，轮询仍是实用基线。它保留熟悉的请求边界，响应允许时还能利用缓存。如果主要是服务端向客户端推送，Server-Sent Events 也值得考虑。先比较业务需要的延迟与长期维护成本，再决定是否持有会话；连接持续打开，并不自动意味着整体方案更简单。
在应用设计中，应先认证再授予访问权限，检查浏览器来源，对订阅进行授权，并明确凭据过期或权限变化后的处理方式。重连应重新取得有效授权，再重建订阅。应用可以在消息中携带授权信息，但 WebSocket 不会替应用制定这些规则。握手成功也不保证用户在整个会话期间永远具有相同权限，必须把身份的生命周期与连接的生命周期分别考虑。
重新连接本身既不会自动恢复订阅，也不会补发遗漏事件。如果业务不能接受缺口，应设计应用序号、恢复游标或重新读取快照，让客户端知道从哪里继续。不要把旧连接上的本地状态当作新连接天然继承的状态；重连落到另一台服务节点时，这个假设尤其容易暴露问题。恢复动作应有明确起点与可重复的结果。
让会话保持健康
每条连接的发送队列都应有上限，并监测慢消费者。根据消息含义，可以合并已被新值取代的更新、降低发送频率，或者断开持续落后的订阅者；接收侧也要有相应限制。否则，对端即使仍在线，也可能耗尽服务端内存，或让客户端界面无法及时处理积压。背压是容量问题，心跳成功不能证明应用消息正在被足够快地消费。两种检查互相补充，不能互相替代。
协议层 Ping/Pong 是控制帧，不能用裸的 ping 字符串代替。浏览器 JavaScript 的 WebSocket API 不直接开放这些控制帧；应用层心跳则是另外约定的消息交换。客户端发送的帧需要掩码，服务端发送的帧不加掩码。控制帧不得分片，载荷最多一百二十五字节；响应 Ping 的 Pong 应回显相同载荷。应用可以记录一个尚未得到回应的 Ping 及其截止时间，其他载荷或主动发送的 Pong 不必满足这次应用检查。RFC 6455，第 5.1 与 5.5 节。
坏掉的样子
安静的连接可能触发中间设备的空闲超时，但不存在所有代理统一使用的六十秒上限，也没有协议强制规定连续错过两次 Pong 才能断开。心跳间隔与截止时间应依据真实链路和服务目标选择。没有尚待回应的 Ping 时，一次接收超时不能证明漏收了 Pong；即使截止时间已过，也只说明没有及时观察到响应，不能由此确定是哪一台机器或哪一段网络损坏。
服务重启后，所有客户端立即同步重连，可能使认证与订阅处理再次过载。应使用有上限、带随机抖动的指数退避，释放旧会话的资源，并让恢复过程可以重复执行。代理空闲策略、集群切换和大规模重连都需要另外验证，不能从下面的小练习推导出生产系统已经具备这些能力。
快速验证
把完整程序保存为 ws_session_probe.py。无套接字的命令是 python3 ws_session_probe.py –offline，只检查解析器与状态逻辑。在已明确授权的隔离本地环境中，再运行 python3 ws_session_probe.py –loopback；它仅在自己的随机 127.0.0.1 端口上执行十六个有限场景，不安装依赖，也不连接外部服务。
"""Finite HTTP/1.1 WebSocket control-frame fixture, not a production server.
No sockets are created on import or with --offline. --loopback is explicit.
"""
import argparse
import base64
import binascii
import hashlib
import io
import os
import queue
import re
import socket
import threading
import time
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
VECTOR_KEY = "dGhlIHNhbXBsZSBub25jZQ=="
VECTOR_ACCEPT = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
HOST = "127.0.0.1"
IO_SECONDS = 3.0
PONG_SECONDS = 0.75 # Exercise policy, not an RFC or deployment default.
MAX_HEADER = 8192
MAX_FIELDS = 32
TOKEN = r"[!#$%&'*+.^_`|~0-9A-Za-z-]+"
def check(ok, message):
if not ok:
raise AssertionError(message) # Also active under python -O.
def accept_value(key):
return base64.b64encode(hashlib.sha1((key + GUID).encode("ascii")).digest()).decode()
def parse_head(raw):
if len(raw) > MAX_HEADER or not raw.endswith(b"\r\n\r\n"):
raise ValueError("header limit or termination")
lines = raw[:-4].decode("ascii").split("\r\n")
if len(lines) - 1 > MAX_FIELDS:
raise ValueError("field limit")
headers = {}
for line in lines[1:]:
name, colon, value = line.partition(":")
if not colon or not re.fullmatch(TOKEN, name):
raise ValueError("field syntax")
value = value.strip(" \t")
if any(ord(c) < 32 and c != "\t" or ord(c) == 127 for c in value):
raise ValueError("field control byte")
name = name.lower()
if name in headers:
if name not in ("connection", "upgrade"):
raise ValueError("duplicate singleton (fixture policy)")
value = headers[name] + "," + value
headers[name] = value
return lines[0], headers
def list_values(value, protocol=False):
grammar = TOKEN + (r"(?:/" + TOKEN + r")?" if protocol else "")
parts = [part.strip(" \t") for part in value.split(",") if part.strip(" \t")]
if not parts or any(not re.fullmatch(grammar, part) for part in parts):
raise ValueError("invalid token list")
return {part.lower() for part in parts}
def validate_request(raw, authority):
try:
start, h = parse_head(raw)
if start != "GET /chat HTTP/1.1" or h.get("host") != authority:
return 400, None
if "websocket" not in list_values(h.get("upgrade", ""), protocol=True):
return 400, None
if "upgrade" not in list_values(h.get("connection", "")):
return 400, None
# The fixture has no request-body support and never negotiates extensions.
if "content-length" in h or "transfer-encoding" in h:
return 400, None
key = h.get("sec-websocket-key", "")
if len(base64.b64decode(key, validate=True)) != 16:
return 400, None
version = h.get("sec-websocket-version", "")
if not re.fullmatch(r"0|[1-9][0-9]{0,2}", version) or int(version) > 255:
return 400, None
if version != "13":
return 426, None
return 101, key
except (ValueError, UnicodeError, binascii.Error):
return 400, None
def response(status, key=None):
if status == 101:
return ("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n"
"Connection: Upgrade\r\nSec-WebSocket-Accept: " + accept_value(key)
+ "\r\n\r\n").encode("ascii")
# An empty ordinary HTTP error response may use Content-Length.
extra = "Sec-WebSocket-Version: 13\r\nUpgrade: websocket\r\n" if status == 426 else ""
reason = "Upgrade Required" if status == 426 else "Bad Request"
connection = "Upgrade, close" if status == 426 else "close"
return (f"HTTP/1.1 {status} {reason}\r\n{extra}"
f"Connection: {connection}\r\nContent-Length: 0\r\n\r\n").encode("ascii")
def exact(sock, n, deadline):
data = bytearray()
while len(data) < n:
remaining = deadline - time.monotonic()
if remaining <= 0:
raise TimeoutError("absolute receive deadline")
sock.settimeout(remaining)
chunk = sock.recv(n - len(data))
if not chunk:
raise EOFError("peer closed early")
data.extend(chunk)
return bytes(data)
def receive_head(sock):
deadline = time.monotonic() + IO_SECONDS
data = bytearray()
while not data.endswith(b"\r\n\r\n"):
if len(data) >= MAX_HEADER:
raise ValueError("header limit")
data.extend(exact(sock, 1, deadline)) # Never consumes following frames.
return bytes(data)
def send(sock, data):
sock.settimeout(IO_SECONDS)
sock.sendall(data)
def frame(opcode, payload, masked=False):
if opcode not in (8, 9, 10) or len(payload) > 125:
raise ValueError("fixture supports short control frames only")
mask = os.urandom(4) if masked else b""
body = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) if masked else payload
return bytes((0x80 | opcode, (0x80 if masked else 0) | len(payload))) + mask + body
def parse_control(read, expect_masked):
"""read(n) must return exactly n bytes or raise, as exact() does."""
first, second = read(2)
opcode, size = first & 15, second & 127
if first & 0x70 or not first & 0x80 or opcode not in (8, 9, 10) or size > 125:
raise ValueError("invalid or unsupported control frame")
if bool(second & 0x80) != expect_masked:
raise ValueError("wrong mask direction")
mask = read(4) if expect_masked else b""
data = read(size)
payload = bytes(b ^ mask[i % 4] for i, b in enumerate(data)) if mask else data
# Only our empty/1000/1002/1008 Close bodies are within this fixture's scope.
if opcode == 8 and payload not in (b"", b"\x03\xe8", b"\x03\xea", b"\x03\xf0"):
raise ValueError("Close body outside fixture scope")
return opcode, payload
def receive_control(sock, masked, deadline=None):
end = time.monotonic() + IO_SECONDS if deadline is None else deadline
return parse_control(lambda n: exact(sock, n, end), masked)
class Heartbeat:
def __init__(self):
self.outstandingPing = None
self.deadline = None
def sent(self, payload, now):
check(self.outstandingPing is None, "only one outstanding Ping")
self.outstandingPing, self.deadline = payload, now + PONG_SECONDS
def pong(self, payload, now):
if self.outstandingPing is not None and payload == self.outstandingPing and now < self.deadline:
self.outstandingPing, self.deadline = None, None
return True
return False
def expired(self, now):
return self.outstandingPing is not None and now >= self.deadline
def server_session(conn):
try:
opcode, payload = receive_control(conn, True)
except ValueError:
send(conn, frame(8, b"\x03\xea")) # 1002 protocol error.
return "invalid-frame-refused"
check((opcode, payload) == (9, b"client-ping"), "masked client Ping")
send(conn, frame(10, payload))
hb = Heartbeat()
nonce = os.urandom(8)
send(conn, frame(9, nonce))
hb.sent(nonce, time.monotonic()) # Record only after successful Ping send.
mismatches = 0
for _ in range(4): # A malicious/unexpected frame stream cannot loop forever.
try:
opcode, payload = receive_control(conn, True, hb.deadline)
except TimeoutError:
check(hb.expired(time.monotonic()), "timeout needs outstanding Ping")
send(conn, frame(8, b"\x03\xf0")) # 1008: this exercise's policy.
check(receive_control(conn, True) == (8, b"\x03\xf0"), "Close reply")
return "missing-pong-expired"
check(opcode == 10, "expected Pong in bounded heartbeat scenario")
if hb.pong(payload, time.monotonic()):
check(mismatches == 1 and hb.outstandingPing is None, "matched Pong clears state")
send(conn, frame(8, b"\x03\xe8"))
check(receive_control(conn, True) == (8, b"\x03\xe8"), "normal Close reply")
return "mismatch-ignored-match-cleared"
mismatches += 1
raise AssertionError("heartbeat frame budget exhausted")
def request(authority, changes=None):
h = {"Host": authority, "Upgrade": "WebSocket", "Connection": "keep-alive, UpGrAdE",
"Sec-WebSocket-Key": base64.b64encode(os.urandom(16)).decode(), "Sec-WebSocket-Version": "13"}
for key, value in (changes or {}).items():
if value is None:
h.pop(key, None)
else:
h[key] = value
return ("GET /chat HTTP/1.1\r\n" + "".join(f"{k}: {v}\r\n" for k, v in h.items())
+ "\r\n").encode("ascii")
BAD_REQUESTS = [
("connection-substring", {"Connection": "notupgrade"}, 400),
("connection-syntax", {"Connection": "Upgrade;bad"}, 400),
("upgrade-substring", {"Upgrade": "xwebsocket"}, 400),
("upgrade-syntax", {"Upgrade": "websocket;bad"}, 400),
("missing-version", {"Sec-WebSocket-Version": None}, 400),
("malformed-version", {"Sec-WebSocket-Version": "13,12"}, 400),
("unsupported-version", {"Sec-WebSocket-Version": "12"}, 426),
("missing-key", {"Sec-WebSocket-Key": None}, 400),
("malformed-key", {"Sec-WebSocket-Key": "!not-base64!"}, 400),
("short-key", {"Sec-WebSocket-Key": "c2hvcnQ="}, 400),
]
# Invalid two-byte headers are rejected before any body is required/read.
BAD_FRAMES = [
("unmasked-client", b"\x89\x00"),
("fragmented-control", b"\x09\x80"),
("oversized-control", b"\x89\xfe"),
("reserved-bit", b"\xc9\x80"),
]
def offline():
check(accept_value(VECTOR_KEY) == VECTOR_ACCEPT, "RFC accept vector")
for name, changes, expected in BAD_REQUESTS:
check(validate_request(request("127.0.0.1:1", changes), "127.0.0.1:1")[0] == expected, name)
check(validate_request(request("127.0.0.1:1"), "127.0.0.1:1")[0] == 101, "valid tokens")
for masked in (False, True):
reader = io.BytesIO(frame(9, b"hello", masked)).read
check(parse_control(reader, masked) == (9, b"hello"), "frame round trip")
for name, raw in BAD_FRAMES:
try:
parse_control(io.BytesIO(raw).read, True)
except ValueError:
pass
else:
raise AssertionError(name)
hb = Heartbeat()
check(not hb.expired(100), "idle has no missed-Pong deadline")
hb.sent(b"p", 100)
check(not hb.pong(b"wrong", 100.1), "mismatched Pong must not clear")
check(hb.pong(b"p", 100.2) and not hb.expired(200), "matched Pong clears")
hb.sent(b"q", 200)
check(not hb.expired(200.1) and hb.expired(201), "outstanding deadline")
check(not hb.pong(b"q", 201), "late Pong does not revive expired state")
start, h = parse_head(response(101, VECTOR_KEY))
check(start == "HTTP/1.1 101 Switching Protocols", "exact status")
check("content-length" not in h and "transfer-encoding" not in h, "no 101 HTTP body framing")
print("OFFLINE ONLY: parser, vector, frame codec and state checks passed; sockets unexecuted")
def loopback():
cases = [(name, changes, status, None) for name, changes, status in BAD_REQUESTS]
cases += [(name, {}, 101, raw) for name, raw in BAD_FRAMES]
cases += [("matched-pong", {}, 101, None), ("missing-pong", {}, 101, None)]
results = queue.Queue()
stop = threading.Event()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
listener.bind((HOST, 0)) # OS-selected port; never a fixed or external service.
listener.listen(1)
listener.settimeout(IO_SECONDS)
authority = f"{HOST}:{listener.getsockname()[1]}"
print(f"OWNED_LOOPBACK {authority}", flush=True)
def worker():
try:
for _ in cases:
if stop.is_set():
return
conn, peer = listener.accept()
with conn:
check(peer[0] == HOST, "loopback peer")
status, key = validate_request(receive_head(conn), authority)
send(conn, response(status, key))
results.put(server_session(conn) if status == 101 else f"http-{status}")
except Exception as error:
results.put(error)
thread = threading.Thread(target=worker, name="owned-ws-fixture", daemon=False)
thread.start()
try:
for name, changes, expected, bad_frame in cases:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
client.settimeout(IO_SECONDS)
client.connect(listener.getsockname()) # No DNS or external request.
raw = request(authority, changes)
send(client, raw)
start, h = parse_head(receive_head(client))
check(start.split(" ")[1] == str(expected), name + " status")
if expected != 101:
check("sec-websocket-accept" not in h, name + " no upgrade")
if expected == 426:
check(h.get("sec-websocket-version") == "13", "advertise supported version")
wanted = f"http-{expected}"
else:
key = parse_head(raw)[1]["sec-websocket-key"]
check(h.get("sec-websocket-accept") == accept_value(key), "accept matches nonce")
check(list_values(h.get("upgrade", ""), True) == {"websocket"}, "response upgrade")
check(list_values(h.get("connection", "")) == {"upgrade"}, "response connection")
check("content-length" not in h and "transfer-encoding" not in h, "101 framing")
if bad_frame is not None:
send(client, bad_frame)
check(receive_control(client, False) == (8, b"\x03\xea"), "1002 rejection")
wanted = "invalid-frame-refused"
else:
send(client, frame(9, b"client-ping", True))
check(receive_control(client, False) == (10, b"client-ping"), "echoed Pong")
opcode, nonce = receive_control(client, False)
check(opcode == 9 and len(nonce) == 8, "server Ping received")
if name == "matched-pong":
send(client, frame(10, b"wrong", True))
send(client, frame(10, nonce, True))
close = b"\x03\xe8"
wanted = "mismatch-ignored-match-cleared"
else:
# Keep the TCP socket open but deliberately send no Pong.
close = b"\x03\xf0"
wanted = "missing-pong-expired"
check(receive_control(client, False) == (8, close), "server Close")
send(client, frame(8, close, True))
outcome = results.get(timeout=2 * IO_SECONDS)
if isinstance(outcome, Exception):
raise outcome
check(outcome == wanted, name + " server outcome")
client.settimeout(IO_SECONDS)
check(client.recv(1) == b"", name + " owned connection closed")
print("CASE_OK", name, outcome, flush=True)
finally:
stop.set()
# Every blocking operation is bounded; failure closes the active client.
thread.join(timeout=3 * IO_SECONDS + PONG_SECONDS)
check(not thread.is_alive(), "owned worker failed to stop")
check(listener.fileno() == -1, "owned listener closed")
print(f"PASS LOOPBACK: {len(cases)} cases; owned sockets closed; worker joined")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
mode = parser.add_mutually_exclusive_group(required=True)
mode.add_argument("--offline", action="store_true")
mode.add_argument("--loopback", action="store_true")
args = parser.parse_args()
offline() if args.offline else loopback()
实际运行的预期结果包括：十种无效握手被拒绝，四种无效控制帧被拒绝，一个错误载荷的 Pong 不清除等待状态、随后匹配的 Pong 清除状态，以及真实 Ping 发出后故意不回 Pong 而触发截止时间。全部成功才输出 PASS LOOPBACK，并确认自己创建的套接字已关闭、工作线程已回收。任一检查失败都会抛异常，头部、帧、读取与写入均有明确上限。
历史执行记录：这份完全相同的共享程序于 2026-10-17 在 macOS 26.5.1 arm64、Python 3.14.4 上完成一次受监督的十六场景试验。两种语言中的程序字节相同；这是一次共享试验，不是分别执行了两份语言版本。保留的输出和清理记录只支持这里列明的场景，本次完整源码复核没有再次执行套接字或浏览器测试。这是具有狭窄 HTTP 解析范围和有限控制帧能力的教学夹具，不是完整 WebSocket 服务端。TLS、浏览器、HTTP/2、数据分片、压缩、代理行为、负载和生产认证均不在本次覆盖范围内，不能因为程序最终打印成功就把这些能力一并视为通过。
延伸阅读
来源
- MDN：编写 WebSocket 客户端，API 与生命周期，已核验 2026-10-19。
- RFC 6455：WebSocket 协议，握手与分帧，已核验 2026-10-19。
- MDN：服务器发送事件(SSE)，更简单的单向推送替代，已核验 2026-10-19。
- RFC 9110：HTTP 语义，101 不得携带 Content-Length，已核验 2026-10-19。
- RFC 9112：HTTP/1.1，101 不得携带 Transfer-Encoding，已核验 2026-10-19。
- RFC 8441：HTTP/2 上的 WebSocket，使用扩展 CONNECT 而非 HTTP/1.1 升级，已核验 2026-10-19。
- WHATWG WebSocket 标准，浏览器 API 不开放协议 Ping/Pong，已核验 2026-10-19。
