记录生活
所有文章 分类 标签 60s Wallpaper Chat RSS Casual
记录生活

Contents

Websocket

   6855 words   14 minutes 

参考资料

简介

一个即时聊天、多人协作的 web 应用, 其客户端和服务器之间常见的 双向数据交换方式 有：

  • 短轮询
  • 长轮询
  • SSE（Server-Sent-Events）

以上方法存在着 效率、网络带宽利用率 等等的问题

WebSocket 是一种 在单个 TCP 连接上实现全双工（双向）通信 的协议
它由 RFC 6455（2011 年）正式定义, 旨在解决传统 HTTP 请求/响应模型在实时交互场景下的低效问题

为什么需要 WebSocket

HTTP 协议有一个缺陷：通信只能由客户端发起, 服务器无法主动向客户端推送信息, 即单向请求
如果服务器有连续的状态变化, 客户端要获知就非常麻烦
而 轮询 效率低, 且浪费资源（客户端每隔一段时间就发出一个询问）
WebSocket 应运而生

  • 常见使用场景
    场景 典型需求 WebSocket 优势
    即时聊天 双向低延迟文本传输 实时推送、低开销
    多人在线游戏 高频率的二进制状态更新 二进制帧、低延迟
    实时仪表盘/监控 推送监控数据、图表刷新 服务器主动推送
    协同编辑（文档、代码） 同步编辑操作 多用户即时广播
    物联网（IoT） 设备状态上报、指令下发 支持子协议（如 MQTT）
    金融行情 高频行情推送 减少网络延迟、带宽占用低
    推送通知 服务器主动下发消息 取代轮询/长轮询, 提高效率

WebSocket 特点

  • 全双工：客户端和服务器可以同时主动发送数据, 而不需要像 AJAX 那样每次都发起新的 HTTP 请求
  • 低开销：建立一次连接后, 后续的数据帧只包含很少的协议头部, 避免了重复的 TCP/HTTP 握手开销
  • 实时性：适用于聊天、在线游戏、股票行情、物联网数据推送等需要 低延迟 的场景
  • 没有同源限制, 客户端可以与任意服务器通信
  • 协议标识符是 ws（如果加密, 则为 wss）, 服务器网址就是 URL, 例如：wss://echo.websocket.org
对比维度 HTTP（请求/响应） WebSocket
通信模式 单向（客户端发起请求, 服务器返回响应） 双向（任意一端都可主动发送）
连接复用 每次请求都要建立 TCP 连接（或使用 Keep‑Alive 复用, 但仍是请求/响应） 建立一次后保持长连接, 持续通信
协议开销 每次请求都携带完整的 HTTP 头部 只在握手阶段有 HTTP 头部, 后续帧仅几字节
实时性 轮询/长轮询需要额外延迟 立即推送, 毫秒级延迟
安全 支持 HTTP/HTTPS 支持 ws://（明文） 和 wss://（TLS）

建立连接的握手过程

客户端

使用 HTTP/1.1 的 Upgrade 机制进行一次“升级”握手
客户端一旦发送了连接握手请求, 就必须等待服务器的响应

客户端连接握手请求的 header 部分字段：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 必须是 GET 方法, HTTP 版本至少是 1.1
GET /chat HTTP/1.1
Host: example.com:8080
# Upgrade 值必须是 websocket, 不区分大小写
Upgrade: websocket
# Connection 值必须是 Upgrade, 不区分大小写
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Sec-WebSocket-Protocol: chat, superchat   # 可选, 子协议
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits

请求 URI 格式
ws://host:port/path?query / wss://host:port/path?query
端口号和查询值是可选的

Sec-WebSocket-Key 是客户端为本次建连随机生成的 16 字节的 base64 编码的字符串
Sec-WebSocket-Version 是客户端拟使用协议版本号, 值必须为 13
Sec-WebSocket-Protocol 是客户端支持的一个或多个以逗号分割的子协议, 按优先级排序
Sec-WebSocket-Extensions 客户端拟使用协议扩展, 例如多路复用扩展、压缩扩展

服务器端

服务器收到客户端的连接握手请求时, 必须进行回复
返回 101 Switching Protocols 响应

回复 header 部分字段如下：

1
2
3
4
5
6
7
8
9

HTTP/1.1 101 Switching Protocols
# Upgrade 值必须是 websocket, 不区分大小写
Upgrade: websocket
# Connection 值必须是 Upgrade, 不区分大小写
Connection: Upgrade
Status Code: 101 Switching Protocols
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
# Sec-WebSocket-Protocol 服务器拟使用的协议, 从客户端发送的里面选择, 若服务器都不支持, 则值为空
Sec-WebSocket-Protocol: chat

HTTP/1.1 101 Switching Protocols 表示接受客户端的请求
若服务器想要停止处理握手, 可以返回 401 等类型的错误代码的 HTTP 响应

服务器接受客户端的请求后, 先将客户端请求头的 Sec-WebSocket-Key 值与 全局唯一标识 258EAFA5-E914-47DA-95CA-C5AB0DC85B11 拼接
然后对拼接的字符串进行 SHA-1 哈希, 再进行 base64-encoded, 最终得到 Sec-WebSocket-Accept

消息传输

建连握手成功后, 连接从 HTTP 切换为 WebSocket
客户端和服务器就可以开始数据传输了, 进入帧（frame）层的通信

WebSocket 帧格式

WebSocket 把每一条消息切分为 帧, 帧的二进制结构如下

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

 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 bit, conditional)  |
|N|V|V|V|       |S|             |    (if payload len==126/127)  |
| |1|2|3|       |K|             |                               |
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|    Extended payload length continued, if payload len==127     |
+ - - - - - - - - - - - - - - - +-------------------------------+
|                               | Masking-key, if MASK set to 1 |
+-------------------------------+-------------------------------+
|    Masking-key (continued)    |        Payload Data           |
+-------------------------------- - - - - - - - - - - - - - - - +
|                    Payload Data continued ...                 |
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|                    Payload Data continued ...                 |
+---------------------------------------------------------------+

  • FIN

    1 bit, 表示是否是一条消息的最后一帧
    1 表示结束, 0 表示还有后续帧

  • RSV1, RSV3, RSV3

    分别为 1 bit, 扩展功能未使用的情况下, 默认为 0
    可用于扩展如 permessage-deflate

  • opcode

    4 bit, 用于表示帧 payload data 的数据类型
    例如十进制、二进制、文本, 还有连接关闭、心跳帧等等

    0x0 - Continuation 续帧
    0x1 - Text utf-8 文本
    0x2 - Binary 二进制
    0x8 - Close lia 连接关闭
    0x9 - Ping 心跳帧
    0xA - Pong

  • MASK

    1 bit, 是否进行掩码, 是为 1, 否为 0
    客户端发送的帧必须进行掩码, 服务器端发送到可以不用

  • Payload length 和 Extended payload length

    Payload length 为 7 位
    Extended payload length 为 16 或者 64 位
    如果 payload data 的长度 <= 125, 则使用 Payload length 表示其长度
    如果 payload data 的长度 == 126, 则使用 Extended payload length 的后 16 位表示
    如果 payload data 的长度 == 127, 则使用 Extended payload length 的后 64 位表示

  • Masking-key 掩码

    32 位, 是由 客户端生成并发送给服务器 的掩码
    用于在数据传输过程中对数据进行掩码处理：

    为了防止 代理缓存污染 攻击, 掩码必须来自强大的熵源, 不可被预测
    常规的算法：以字节为步长遍历载荷数据, 对于载荷数据的第 i 个字节, 对 4 取模得到 j（只能为 0、1、2、3）
    然后将第 i 个字节与 masking-key 的第 j 个字节做 按位异或 操作（相同为 0, 不同为 1）, 混淆原始数据

    代理缓存污染攻击
    缓存污染攻击是一种安全攻击, 攻击者通过发送 恶意构造的数据包, 使得 代理服务器 错误地将这些数据存储在缓存中
    从而可能导致其他用户接收到错误的数据

  • payload data

    真正的业务数据, 为 文本或者二进制, 分为：扩展数据、应用数据

心跳 / Keep‑Alive（Ping‑Pong）

  • Ping：服务器（或客户端）发送 opcode=0x9 的 Ping 帧, 携带可选负载。
  • Pong：对方必须在收到 Ping 后 立刻回送 Pong 帧（opcode=0xA）, 负载必须相同（若有）。

作用：

  1. 防止 NAT/防火墙在长时间无数据时关闭连接。
  2. 检测对端是否仍然活跃。
  3. 在 ws（Node）等库中可以配置 heartbeat 机制（如每 30 s 发送一次 Ping）。

消息分片

将概念上的一条消息通过多个数据帧发送
允许发送未知大小的消息, 而不必缓冲整条消息
并且可以结合多路复用协议的扩展, 实现分割消息为更小的分段以共享输出通道

要求分片数据帧 按顺序 发送到另一端

断连握手

客户端和服务器都可以发送包含指定控制序列的 Close 控制帧
一方发送出关闭控制帧后, 另一方接收到只需要发送一个关闭帧作为响应
然后关闭连接

常用子协议（Subprotocol）

在握手阶段, 客户端可以通过 Sec-WebSocket-Protocol 头部列出它支持的 子协议（例如 chat、json、mqtt）, 服务器从中挑选一个并在响应中返回
子协议本质上是 在同一个连接上协商的业务层协议, 常见的有：

  • graphql-ws（GraphQL 实时订阅）
  • mqtt（IoT 场景）
  • wamp（Web Application Messaging Protocol）
  • json-rpc（JSON‑RPC over WebSocket）

如果不需要子协议, 直接省略该头部即可

安全

  • ws://

    明文协议, 等同于普通 TCP, 数据在网络上未加密
    适用于局域网或安全已受信的环境

  • wss://

    在 TLS（HTTPS）之上运行的 WebSocket, 数据在传输过程中经过加密
    生产环境、跨公网通信、涉及敏感信息时强烈推荐使用。

TLS 握手 将在原始 TCP 握手完成后进行, 随后进入 WebSocket 握手阶段
多数现代浏览器仅在 https:// 页面中允许 wss:// 连接, 以防混合内容（mixed content）安全风险

常见错误码（Close Code）

码 含义 参考
1000 正常关闭（Normal Closure） 正常业务结束
1001 因端点离开（Going Away） 服务器关闭、页面刷新
1002 协议错误（Protocol error） 握手或帧格式错误
1003 不接受的数据类型（Unsupported Data） 收到不支持的二进制/文本
1006 非法关闭（Abnormal Closure） （不在帧中出现） 连接异常断开
1007 无效负载数据（Invalid payload data） 例如不是有效 UTF‑8
1011 服务器内部错误（Server error） 服务器内部异常
1015 TLS 握手失败 （仅在 TLS 中出现） TLS 失败或证书错误

建议：在业务层捕获关闭事件, 并依据返回码执行相应的重连或清理逻辑。

性能与扩展

场景 优化点
高并发（大量连接） 使用 epoll/kqueue 事件模型（Node.js、Go、Rust）或 Nginx + ngx_http_websocket_module 做负载均衡。
大消息（文件、二进制） 开启 permessage-deflate（压缩）或自行切分成更小的帧；使用 binaryType = 'arraybuffer' 减少内存拷贝。
分布式广播 使用 消息队列（Kafka、Redis Pub/Sub）配合多实例 WebSocket 服务器实现全局广播。
安全 使用 JWT 在握手时通过 Sec-WebSocket-Protocol 传递 token, 或在 HTTP 升级请求里使用 Cookie/Authorization Header。
流控 在服务器端对每个连接设定 发送速率限制, 防止单个客户端占用过多带宽。
横向扩展 将 WebSocket 服务器放在 容器/微服务 中, 配合 Ingress（K8s）或 云负载均衡 实现弹性伸缩。

与其他实时技术的对比

技术 双向通信 传输层 编码/协议 典型场景
WebSocket ✅（全双工） TCP 自定义二进制/文本帧 聊天、游戏、实时监控
Server‑Sent Events (SSE) ❌（服务器 → 客户端） HTTP/1.1（长连接） 纯文本流（UTF‑8） 实时日志、新闻推送
Long Polling ✅（伪双向） HTTP 标准请求/响应 兼容旧浏览器
HTTP/2 Push ✅（单向） HTTP/2 二进制帧（SETTINGS, PUSH_PROMISE） 资源预加载
gRPC‑Web ✅（双向） HTTP/2 protobuf 微服务间的高效 RPC
WebRTC DataChannel ✅（双向+P2P） UDP + SCTP 任意二进制 浏览器点对点文件传输、多人游戏

速查表

检查项 一句话记忆
握手 Key + Magic → SHA-1 → Base64, 返回 101
掩码 浏览器 → 服务器必须掩码, 方向反了直接 1002
心跳 30 s 一次 Ping, 无 Pong 触发重连
关闭码 1000 正常, 1006 异常, 1011 服务端崩溃
子协议 前后端名单必须交集非空, 否则握手失败
生产地址 一律 wss://, 配 TLS 1.3 + ALPN=h2
高并发 单进程 1 万连接 ≈ 240 MB 内存（epoll + permessage-deflate 关）
断线重连 指数退避：1 s → 2 s → 4 s … 上限 30 s
广播 本地集合适用于单机, 多机用 Redis Pub/Sub
安全头 Content-Security-Policy: connect-src wss://*.example.com

示例

前端 JS

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <!-- <script src="https://code.jquery.com/jquery-3.5.1.js"></script> -->
    <title>Document</title>
  </head>

  <body>
    <div style="display: flex;">
      <ul id="content"></ul>
      <form class="form">
        <input
          type="text"
          placeholder="请输入发送的消息"
          class="message"
          id="message"
        />
        <input type="button" value="连接" id="connect" class="connect" />
        <input type="button" value="发送" id="send" class="send" />
      </form>
    </div>
    <script type="text/javascript">
      let oUl = document.getElementById("content");
      let oMsg = document.getElementById("message");
      let oConnect = document.getElementById("connect");
      let oSend = document.getElementById("send");
      let websocket = null;
      oConnect.onclick = function () {
        // 创建 WebSocket 连接
        websocket = new WebSocket("ws://127.0.0.1:10083");

        websocket.onopen = function () {
          oUl.innerHTML += "<li>客户端已连接</li>";
        };

        websocket.onmessage = function (event) {
          oUl.innerHTML += "<li>" + event.data + "</li>";
        };

        websocket.onclose = function () {
          oUl.innerHTML += "<li>客户端已断开连接</li>";
        };

        websocket.onerror = function (evt) {
          oUl.innerHTML += "<li>" + evt.data + "</li>";
        };
      };
      oSend.onclick = function () {
        if (websocket) {
          websocket.send(oMsg.value);
        }
      };

      // 发送二进制数据（ArrayBuffer / Blob）
      function sendBinary(data) {
        if (socket.readyState === WebSocket.OPEN) {
          socket.send(data); // data 可以是 Uint8Array, ArrayBuffer, Blob
        } else {
          console.warn("连接未打开, 无法发送");
        }
      }
    </script>
  </body>
</html>

  • 常用 API

    方法 / 属性 说明
    new WebSocket(url, protocols?) url 必须以 ws:// 或 wss:// 开头；protocols 可选, 用于子协议协商。
    socket.readyState 0（CONNECTING）、1（OPEN）、2（CLOSING）、3（CLOSED）。
    socket.send(data) data 可以是 String（文本帧）或 Blob/ArrayBuffer/TypedArray（二进制帧）。
    socket.close([code, reason]) 主动关闭连接, 可提供关闭码（如 1000 正常关闭）和原因。
    socket.binaryType 默认为 'blob', 可改为 'arraybuffer' 控制收到二进制时的类型。

  • 处理断线重连（Reconnection）

    WebSocket 本身不提供自动重连, 需要在业务层自行实现。常见策略：

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

    let socket;
let reconnectAttempts = 0;
const maxAttempts = 5;

function connect() {
  socket = new WebSocket("wss://example.com/ws");

  socket.addEventListener("open", () => {
    console.log("已连接");
    reconnectAttempts = 0; // 成功后重置计数
  });

  socket.addEventListener("close", () => {
    if (reconnectAttempts < maxAttempts) {
      const delay = Math.pow(2, reconnectAttempts) * 1000; // 指数退避
      console.log(`将在 ${delay}ms 后重连…`);
      setTimeout(() => {
        reconnectAttempts++;
        connect();
      }, delay);
    } else {
      console.warn("已超过最大重连次数");
    }
  });

  // 其它事件（message、error）同理
}
connect();
    • 指数退避（Exponential Backoff）可以防止在网络故障时产生洪水式请求。
    • 保留未发送的消息：在断线期间缓存待发送的数据, 待重连成功后再发送。

后端 Node.js

1

npm install ws   # 安装库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// server.js
const http = require("http");
const WebSocket = require("ws");

// 创建普通 HTTP 服务器（用于处理升级请求）, 也可以直接使用 https.createServer()
const server = http.createServer((req, res) => {
  res.writeHead(200);
  res.end("WebSocket Server");
});

// 在同一端口上挂载 WebSocket
const wss = new WebSocket.Server({ server });

wss.on("connection", (ws, req) => {
  // 可通过 req.headers 获取握手信息（如 Sec-WebSocket-Protocol）
  console.log("客户端已连接, 地址：", req.socket.remoteAddress);

  // 发送欢迎消息（文本）
  ws.send("Welcome!");

  // 监听消息
  ws.on("message", (message) => {
    console.log("收到客户端消息：", message);
    // 回显（demo）
    ws.send(`Server echo: ${message}`);
  });

  // 处理错误
  ws.on("error", (err) => {
    console.error("WebSocket error:", err);
  });

  // 处理关闭
  ws.on("close", (code, reason) => {
    console.log(`连接关闭, 代码=${code}, 原因=${reason}`);
  });
});

// 监听端口
server.listen(8080, () => {
  console.log("HTTP + WebSocket server listening on http://localhost:8080");
});

  • 常见配置项

    参数 说明
    { server } 将 WebSocket.Server “挂载”到已有的 HTTP/HTTPS 服务器上。
    port 直接创建独立的 WS 服务器（不依赖 HTTP）。
    verifyClient 用于在握手阶段进行身份验证（返回 true/false 或调用回调）。
    handleProtocols 用于子协议协商, 自定义返回子协议名称。
    perMessageDeflate 启用/禁用 permessage-deflate（压缩扩展）, 默认开启。

    生产环境推荐：使用 HTTPS + wss, 并在握手阶段进行 身份认证（如 JWT、Session Cookie）, 防止未授权的连接。

后端 Python

1

pip install websockets

websockets 是一个基于 asyncio 的库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

import asyncio
import websockets
import logging
import json
import random


logging.basicConfig(level=logging.INFO)
connected_clients = set()
choices = ["hello", "world", "hi", "yes", "good"]


async def some_dealwith(websocket):
    pass


# 简单处理客户端传来的消息
async def recv_msg(websocket, send_event):
    while True:
        try:
            recv_text = await websocket.recv()
            logging.info("client: " + recv_text)
            send_event.set()
        except Exception as e:
            logging.error(e)
            break


# 简单发送数据
async def send_msg(websocket, send_event):
    while True:
        try:
            # 等待 recv_msg 执行后, send_event 被触发
            await send_event.wait()
            send_event.clear()  # 重置事件
            response_text = ''.join(['server: ', random.choice(choices)])
            await websocket.send(response_text)
        except Exception as e:
            logging.error(e)
            break


# 服务器主逻辑
async def main_logic(websocket, path):
    await some_dealwith(websocket)
    connected_clients.add(websocket)
    send_event = asyncio.Event()
    try:
        # 使用 asyncio.gather 并发运行 recv_msg 和 send_msg 函数。
        await asyncio.gather(
            recv_msg(websocket, send_event), send_msg(websocket, send_event)
        )
    except websockets.ConnectionClosed as e:
        print(f"closed {e}")
    finally:
        connected_clients.remove(websocket)


# 创建服务器
start_server = websockets.serve(main_logic, "localhost", 10083)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

基本功能

  • 创建服务器

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

    # server.py
import asyncio
import websockets

async def handler(websocket, path):
    print(f"Client connected: {path}")
    await websocket.send("Welcome from Python server!")

    async for message in websocket:
        print(f"Received: {message}")
        await websocket.send(f"Echo: {message}")

start_server = websockets.serve(handler, "localhost", 8765, ssl=None)  # 若要 wss, 需要提供 SSLContext

asyncio.get_event_loop().run_until_complete(start_server)
print("WebSocket server listening on ws://localhost:8765")
asyncio.get_event_loop().run_forever()

  • 创建客户端

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

    # client.py
import asyncio
import websockets

async def hello():
    async with websockets.connect("ws://localhost:8765") as ws:
        await ws.send("Hello Server")
        reply = await ws.recv()
        print(f"Server replied: {reply}")

asyncio.run(hello())

  • 广播

    将消息发送给所有连接到服务器的客户端 通过设置一个集合, 存放连接的客户端 每次发送消息的时候, 遍历客户端集合进行 send 最后记得将套接字移出集合

  • SSL/TLS 加密

    通过导入 ssl 库对内容进行加密 然后 websockets.serve() 可以接受一个 ssl 参数

  • 自定义协议

    用于扩展 websocket 的功能 通过继承 websockets.WebSocketServerProtocol 自定义类

实际应用

  • 实时聊天应用

    采用上面说的广播机制

  • 实时数据流

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

    import asyncio
import websockets
import random
import time


async def data_stream(websocket, path):
  while True:
    # 模拟数据流
    data = random.randint(1, 100)
    await websocket.send(str(data))
    await asyncio.sleep(1)

start_server = websockets.serve(echo, 'localhost', 6789)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

手动实现 WebSocket 协议

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

# 继承 threading.Thread 类, 重写run方法：https://www.yisu.com/ask/30375634.html
# 创建服务器
# 处理与客户端的握手和数据交互
import socket
import struct
import hashlib
import base64
import threading
from functools import partial


class MyThread(threading.Thread):
    def __init__(self, target):
        super().__init__()
        self._target = target
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def run(self):
        # 通过使用回调函数, 实现 MyThread 类的可重用
        while not self._stop_event.is_set():
            self._target()


# 获取请求头信息
def get_headers(data):
    headers = {}
    data = str(data, encoding="utf-8")
    header, body = data.split("\r\n\r\n", 1)
    header_list = header.split("\r\n")
    for i in header_list:
        i_list = i.split(":", 1)
        if len(i_list) >= 2:
            headers[i_list[0]] = "".join(i_list[1::]).strip()
        else:
            i_list = i.split(" ")
            if i_list and len(i_list) == 3:
                headers["method"] = i_list[0]
                headers["target"] = i_list[1]
                headers["protocol"] = i_list[2]
                print(
                    "请求方法：{}\n请求目标：{}\n协议版本：{}".format(
                        i_list[0], i_list[1], i_list[2]
                    )
                )
    print(headers)
    return headers


# 封装并发送数据到浏览器
def send_msg(conn, msg_bytes):
    # 接收的第一个字节都是 x81 不变
    first_byte = b"\x81"
    length = len(msg_bytes)
    if length < 126:
        first_byte += struct.pack("B", length)
    elif length <= 0xFFFF:
        first_byte += struct.pack("!BH", 126, length)
    else:
        first_byte += struct.pack("!BQ", 127, length)
    msg = first_byte + msg_bytes
    conn.sendall(msg)
    return True


# 接收数据时的解码过程
def parse_payload(payload):
    payload_len = payload[1] & 127
    if payload_len == 126:
        mask = payload[4:8]
        decoded = payload[8:]
    elif payload_len == 127:
        mask = payload[10:14]
        decoded = payload[14:]
    else:
        mask = payload[2:6]
        decoded = payload[6:]
    # 将所有数据全部收集起来, 对所有字符串编码
    bytes_list = bytearray()
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]
        bytes_list.append(chunk)
    body = str(bytes_list, encoding="utf-8")
    return body


# 从浏览器中接收数据
def recv_msg(conn):
    data_recv = conn.recv(8096)
    if data_recv[0:1] == b"\x81":
        data_parse = parse_payload(data_recv)
        return data_parse
    return False


# 建立握手流程并创建handler_msg完成数据收发
def handler_accept(sock):
    while True:
        conn, addr = sock.accept()
        data = conn.recv(8096)
        headers = get_headers(data)
        # 对请求头中的 sec-websocket-key 进行加密
        response_tpl = (
            "HTTP/1.1 101 Switching Protocols\r\n"
            "Upgrade:websocket\r\n"
            "Connection: Upgrade\r\n"
            "Sec-WebSocket-Accept: %s\r\n"
            "WebSocket-Location: ws://%s\r\n\r\n"
        )
        # 加盐操作, 此处是 H5 规范定义好的
        magic_string = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
        if headers.get("Sec-WebSocket-Key"):
            value = headers["Sec-WebSocket-Key"] + magic_string
        # 对数据进行加解密
        ac = base64.b64encode(hashlib.sha1(value.encode("utf-8")).digest())
        response_str = response_tpl % (ac.decode("utf-8"), headers.get("Host"))
        # 相应握手包数据
        conn.sendall(bytes(response_str, encoding="utf-8"))
        msgThread = MyThread(target=partial(handler_msg, conn))
        msgThread.start()


# 主函数,用于实现数据交互
def handler_msg(connect):
    with connect as connect_ptr:
        while True:
            try:
                recv = recv_msg(connect_ptr)
                print("接收数据：{}".format(recv))
                send_msg(connect_ptr, bytes("helio lyshark", encoding="utf-8"))
            except Exception:
                exit(0)


if __name__ == "__main__":
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(("127.0.0.1", 10083))
    sock.listen(5)

    import time

    t1 = MyThread(target=partial(handler_accept, sock))
    t1.start()

    try:
        while True:
            time.sleep(2)
            print("!!!!!")
    except KeyboardInterrupt:
        t1.stop()
        print("stoping...")
    finally:
        t1.join()
        print("stoped.")


# 一开始的情况是
# accept函数的while循环中开启一个线程收发数据
# 然后创建一个线程执行accept函数
# 主线程监听 ctrl+C

小练习

把“手动实现”改成 asyncio + no external lib, 单文件跑通浏览器回声测试, 加深帧边界处理印象
用 Chrome DevTools → Network → WS 过滤, 故意发一个 126 长度帧, 抓包确认 Extended payload length 字段字节序
给 Node 示例加上 verifyClient 做 JWT 校验, 拒绝无 token 连接, 体验“握手阶段即鉴权”

Updated on 2025-12-04
Back | Home