系统编程

Zig WebSocket 与实时通信：服务端推送与帧解析

WebSocket 是全双工实时通信的核心协议。本文系统讲解 WebSocket 握手升级、帧结构二进制解析（FIN/opcode/mask/payload）、Zig 实现高效服务端、心跳 ping/pong 保活、广播与房间模型的并发设计、与现有 HTTP 服务共存方案，以及支持多客户端的完整工程实践。

Plume 2026-10-01 6 分钟阅读 2920 字

引言

WebSocket 协议（RFC 6455）允许客户端与服务端在单一 TCP 连接上建立全双工通信通道，是实时推送、在线协作、消息系统的技术基石。Zig 没有任何运行时开销，配合直接的字节操作，实现 WebSocket 服务端比在高抽象层语言中更高效、更可预测。

本文从 HTTP Upgrade 握手讲起，覆盖 WebSocket 帧结构二进制解析、服务端实现、ping/pong 心跳、广播与房间模型，以及与 std.http.Server 共存的工程模式。

前置：/zig-http-server/（std.http.Server 基础）、/zig-concurrency-atomics/（线程同步与 Mutex）、/zig-async-network/（非阻塞 I/O）。

1. WebSocket 握手：HTTP Upgrade

WebSocket 连接始于标准 HTTP 请求，客户端发送带 Connection: Upgrade 和 Upgrade: websocket 头的 GET 请求，服务端验证后返回 101 Switching Protocols。

1.1 客户端请求关键头

GET /chat HTTP/1.1
Host: example.com
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

1.2 服务端响应（Zig 生成）

const std = @import("std");
const crypto = std.crypto;

// 服务端接收 Key，追加 "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"，做 SHA-1，再 Base64
pub fn generateAcceptKey(
    client_key: []const u8,
    out_buf: []u8,
) ![]const u8 {
    const magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
    var hasher = crypto.hash.Sha1.init(.{});
    hasher.update(client_key);
    hasher.update(magic);
    var digest: [20]u8 = undefined;
    hasher.final(&digest);

    // Base64 编码
    const encoder = std.base64.standard.Encoder;
    const needed = encoder.calcSize(digest.len);
    std.debug.assert(out_buf.len >= needed);
    return encoder.encode(out_buf[0..needed], &digest);
}

响应格式：

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

注意：握手完成后，TCP 连接仍处于打开状态，但通信协议从 HTTP 切换为 WebSocket 二进制帧。

2. WebSocket 帧格式与二进制解析

RFC 6455 定义了统一的帧格式，所有消息都封装在这种帧中传输。

2.1 帧结构

 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|             |   (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 ...                |
+---------------------------------------------------------------+

2.2 关键字段

字段	位宽	说明
FIN	1 bit	1 = 最后一帧，0 = 分片消息中间帧
RSV1-3	3 bit	扩展保留，默认 0
opcode	4 bit	0x1=文本 0x2=二进制 0x8=Close 0x9=Ping 0xA=Pong
MASK	1 bit	客户端→服务端必须=1，服务端→客户端必须=0
payload len	7 bit	0-125 = 直接长度；126 = 后续2字节；127 = 后续8字节
masking key	32 bit	仅 MASK=1 时存在，XOR 解密用

3. Zig 实现帧编解码器

3.1 帧头解析

pub const OpCode = enum(u4) {
    continuation = 0x0,
    text = 0x1,
    binary = 0x2,
    close = 0x8,
    ping = 0x9,
    pong = 0xA,
};

pub const FrameHeader = struct {
    fin: bool,
    opcode: OpCode,
    masked: bool,
    payload_len: u64,
    mask_key: [4]u8,
};

pub fn parseFrameHeader(reader: anytype) !FrameHeader {
    const b0 = try reader.readByte();
    const b1 = try reader.readByte();

    const fin = (b0 & 0x80) != 0;
    const opcode: OpCode = @enumFromInt(b0 & 0x0F);
    const masked = (b1 & 0x80) != 0;

    var payload_len: u64 = @intCast(b1 & 0x7F);
    if (payload_len == 126) {
        // 2 字节扩展长度（大端）
        const hi = try reader.readByte();
        const lo = try reader.readByte();
        payload_len = (@as(u64, hi) << 8) | lo;
    } else if (payload_len == 127) {
        // 8 字节扩展长度
        payload_len = try reader.readInt(u64, .big);
    }

    var mask_key = [_]u8{0, 0, 0, 0};
    if (masked) {
        _ = try reader.readAll(&mask_key);
    }

    return FrameHeader{
        .fin = fin,
        .opcode = opcode,
        .masked = masked,
        .payload_len = payload_len,
        .mask_key = mask_key,
    };
}

3.2 payload 解掩码

客户端发送的数据必须解掩码：

pub fn unmaskPayload(mask: [4]u8, payload: []u8) void {
    for (payload, 0..) |*byte, i| {
        byte.* ^= mask[i % 4];
    }
}

3.3 帧编码（服务端发送）

pub fn writeFrame(
    writer: anytype,
    opcode: OpCode,
    payload: []const u8,
) !void {
    // 第一个字节：FIN=1, RSV=0, opcode
    const b0: u8 = 0x80 | @intFromEnum(opcode);
    try writer.writeByte(b0);

    // 第二个字节：MASK=0, length
    var b1: u8 = 0;
    if (payload.len <= 125) {
        b1 = @intCast(payload.len);
    } else if (payload.len <= 65535) {
        b1 = 126;
    } else {
        b1 = 127;
    }
    try writer.writeByte(b1);

    // 扩展长度
    if (payload.len > 125 and payload.len <= 65535) {
        try writer.writeInt(u16, @intCast(payload.len), .big);
    } else if (payload.len > 65535) {
        try writer.writeInt(u64, payload.len, .big);
    }

    // 服务端发送不掩码
    try writer.writeAll(payload);
}

4. 服务端核心：连接循环

4.1 连接处理器

pub const Conn = struct {
    stream: std.net.Stream,
    allocator: std.mem.Allocator,

    pub fn handle(self: *Conn) !void {
        const reader = self.stream.reader();
        const writer = self.stream.writer();
        var buf: [4096]u8 = undefined;

        while (true) {
            // 解析帧头
            const header = parseFrameHeader(reader) catch |err| {
                // 连接断开或解析错误
                if (err == error.EndOfStream) return;
                return err;
            };

            if (header.payload_len > buf.len) return error.PayloadTooLarge;
            const payload = buf[0..header.payload_len];
            try reader.readNoEof(payload);

            if (header.masked) {
                unmaskPayload(header.mask_key, payload);
            }

            switch (header.opcode) {
                .text, .binary => {
                    // echo 回显（实际业务替换此处）
                    try writeFrame(writer, .text, payload);
                },
                .ping => {
                    // 回复 pong
                    try writeFrame(writer, .pong, payload);
                },
                .close => {
                    // 回复 close 并断开
                    try writeFrame(writer, .close, &[_]u8{});
                    return;
                },
                else => {}, // 忽略其他
            }
        }
    }

    pub fn close(self: *Conn) void {
        self.stream.close();
    }
};

5. 心跳 ping/pong 与超时

WebSocket 协议内置 ping/pong 机制用于保活：

pub const Connection = struct {
    stream: std.net.Stream,
    last_pong_ms: i64,

    // 由服务端主动发送 ping
    pub fn sendPing(self: *Connection) !void {
        var writer = self.stream.writer();
        try writeFrame(writer, .ping, &[_]u8{});
    }

    // 检测超时：若超过 60 秒未收到 pong，断开连接
    pub fn checkTimeout(self: *Connection, now_ms: i64) bool {
        return (now_ms - self.last_pong_ms) > 60_000;
    }
};

// 单独线程/定时器定期遍历所有连接
pub fn heartbeatLoop(connections: *std.ArrayList(Connection)) !void {
    while (true) {
        std.time.sleep(30 * std.time.ns_per_s);
        const now = std.time.milliTimestamp();

        var i: usize = connections.items.len;
        while (i > 0) {
            i -= 1;
            var conn = &connections.items[i];
            if (conn.checkTimeout(now)) {
                conn.stream.close();
                _ = connections.swapRemove(i);
            } else {
                conn.sendPing() catch {
                    conn.stream.close();
                    _ = connections.swapRemove(i);
                };
            }
        }
    }
}

6. 广播与房间模型

6.1 全局广播

pub const Hub = struct {
    mutex: std.Thread.Mutex,
    conns: std.ArrayList(std.net.Stream),

    pub fn broadcast(self: *Hub, message: []const u8) void {
        self.mutex.lock();
        defer self.mutex.unlock();

        for (self.conns.items) |stream| {
            writeFrame(stream.writer(), .text, message) catch {};
            // 发送失败忽略（由心跳线程清理死连接）
        }
    }

    pub fn remove(self: *Hub, stream: std.net.Stream) void {
        self.mutex.lock();
        defer self.mutex.unlock();
        for (self.conns.items, 0..) |s, i| {
            if (s.handle == stream.handle) {
                _ = self.conns.swapRemove(i);
                break;
            }
        }
    }
};

6.2 房间模型

pub const Room = struct {
    name: []const u8,
    mutex: std.Thread.Mutex,
    members: std.ArrayList(std.net.Stream),
};

pub const RoomManager = struct {
    allocator: std.mem.Allocator,
    mutex: std.Thread.Mutex,
    rooms: std.StringHashMap(*Room),

    pub fn join(self: *RoomManager, room_name: []const u8, stream: std.net.Stream) !void {
        self.mutex.lock();
        defer self.mutex.unlock();

        const gop = try self.rooms.getOrPut(room_name);
        if (!gop.found_existing) {
            const room = try self.allocator.create(Room);
            room.* = .{
                .name = try self.allocator.dupe(u8, room_name),
                .mutex = .{},
                .members = std.ArrayList(std.net.Stream).init(self.allocator),
            };
            gop.value_ptr.* = room;
        }
        const room = gop.value_ptr.*;
        room.mutex.lock();
        defer room.mutex.unlock();
        try room.members.append(stream);
    }

    pub fn broadcastTo(self: *RoomManager, room_name: []const u8, message: []const u8) void {
        self.mutex.lock();
        defer self.mutex.unlock();

        if (self.rooms.get(room_name)) |room| {
            room.mutex.lock();
            defer room.mutex.unlock();
            for (room.members.items) |stream| {
                writeFrame(stream.writer(), .text, message) catch {};
            }
        }
    }
};

7. 与 HTTP 服务共存

WebSocket 和 HTTP 可共用同一个端口和事件循环。std.http.Server 处理普通请求，WebSocket Upgrade 请求走握手逻辑后移交帧处理器。

pub fn handleRequest(server: *std.http.Server, hub: *Hub) !void {
    var request = try server.receive();
    const reader = request.reader();
    const writer = request.writer();

    // 检查是否为 WebSocket Upgrade
    const upgrade_header = request.headers.get("Upgrade") orelse "";
    if (std.mem.eql(u8, upgrade_header, "websocket")) {
        // 执行 WebSocket 握手
        const key = request.headers.get("Sec-WebSocket-Key") orelse return error.MissingKey;
        var accept_buf: [32]u8 = undefined;
        const accept = try generateAcceptKey(key, &accept_buf);

        // 写 101 响应
        try writer.writeAll("HTTP/1.1 101 Switching Protocols\r\n");
        try writer.writeAll("Upgrade: websocket\r\n");
        try writer.writeAll("Connection: Upgrade\r\n");
        try writer.print("Sec-WebSocket-Accept: {s}\r\n", .{accept});
        try writer.writeAll("\r\n");

        // 移交连接给 WebSocket 处理器
        const stream = request.connection.stream;
        try hub.conns.append(stream);
        // 启动线程/协程处理帧循环
        return;
    }

    // 普通 HTTP 请求处理
    try writer.writeAll("HTTP/1.1 200 OK\r\nContent-Length: 13\r\n\r\nHello, HTTP!\n");
}

8. 速查表

需求	实现要点
握手响应	SHA-1(Key + Magic) → Base64 → `101 Switching Protocols`
帧头解析	读 2 字节 + 扩展长度 + 可选 mask key
解掩码	`payload[i] ^= mask[i % 4]`
服务端发帧	FIN=1, MASK=0, 直接写 payload
ping/pong	收 ping 回 pong，服务端主动发 ping 保活
广播	`Mutex` + `ArrayList(Stream)` 遍历发送
房间模型	`StringHashMap(*Room)` + 每房间成员列表
HTTP 共存	检测 `Upgrade: websocket` 头，握手后升级

9. 一句话记忆

WebSocket 握手 = SHA-1(Key + Magic) Base64 回 101；帧 = FIN + opcode + MASK + len + key + payload；服务端发不掩码、收客户端要 XOR 解掩码；ping/pong 保活、Mutex + ArrayList 做广播和房间；HTTP Upgrade 检测头后升级，同一端口共存。

延伸阅读

/zig-crypto-security/ — TLS 握手与 wss 安全传输
/zig-debugging-profiling/ — 高并发连接性能分析
/zig-memory-management/ — 长连接内存生命周期管理
[[zig]] — Zig 系统编程专题

// 完整示例：WebSocket 握手 + 帧解析 + echo 服务端

const std = @import("std");
const crypto = std.crypto;

const OpCode = enum(u4) {
    continuation = 0x0,
    text = 0x1,
    binary = 0x2,
    close = 0x8,
    ping = 0x9,
    pong = 0xA,
};

pub fn generateAcceptKey(client_key: []const u8, out: []u8) ![]const u8 {
    const magic = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
    var sha = crypto.hash.Sha1.init(.{});
    sha.update(client_key);
    sha.update(magic);
    var d: [20]u8 = undefined;
    sha.final(&d);
    const e = std.base64.standard.Encoder;
    return e.encode(out[0..e.calcSize(d.len)], &d);
}

pub fn parseFrameHeader(reader: anytype) !struct { fin: bool, opcode: OpCode, masked: bool, len: u64, mask: [4]u8 } {
    const b0 = try reader.readByte();
    const b1 = try reader.readByte();
    const fin = (b0 & 0x80) != 0;
    const opcode: OpCode = @enumFromInt(b0 & 0x0F);
    const masked = (b1 & 0x80) != 0;
    var len: u64 = @intCast(b1 & 0x7F);
    if (len == 126) {
        len = (@as(u64, try reader.readByte()) << 8) | try reader.readByte();
    } else if (len == 127) {
        len = try reader.readInt(u64, .big);
    }
    var mask = [_]u8{0} ** 4;
    if (masked) try reader.readNoEof(&mask);
    return .{ .fin = fin, .opcode = opcode, .masked = masked, .len = len, .mask = mask };
}

pub fn unmask(data: []u8, mask: [4]u8) void {
    for (data, 0..) |*b, i| b.* ^= mask[i % 4];
}

pub fn writeFrame(writer: anytype, opcode: OpCode, data: []const u8) !void {
    try writer.writeByte(0x80 | @intFromEnum(opcode));
    if (data.len <= 125) {
        try writer.writeByte(@intCast(data.len));
    } else if (data.len <= 65535) {
        try writer.writeByte(126);
        try writer.writeInt(u16, @intCast(data.len), .big);
    } else {
        try writer.writeByte(127);
        try writer.writeInt(u64, data.len, .big);
    }
    try writer.writeAll(data);
}

pub fn handleConnection(stream: std.net.Stream) !void {
    var buf: [1024]u8 = undefined;
    const n = try stream.read(&buf);
    const req = buf[0..n];

    // 简易解析 Sec-WebSocket-Key
    const key_prefix = "Sec-WebSocket-Key: ";
    var key_start: ?usize = null;
    var i: usize = 0;
    while (i + key_prefix.len <= req.len) : (i += 1) {
        if (std.mem.eql(u8, req[i..][0..key_prefix.len], key_prefix)) {
            key_start = i + key_prefix.len;
            break;
        }
    }
    if (key_start == null) return error.NoKey;
    const key = req[key_start.?..][0..24]; // Base64 24 字符

    var accept: [32]u8 = undefined;
    const a = try generateAcceptKey(key, &accept);

    const response = try std.fmt.bufPrint(&buf, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {s}\r\n\r\n", .{a});
    try stream.writeAll(response);

    // WebSocket 帧循环（echo）
    const reader = stream.reader();
    const writer = stream.writer();
    var payload_buf: [4096]u8 = undefined;
    while (true) {
        const h = parseFrameHeader(reader) catch break;
        if (h.len > payload_buf.len) break;
        const payload = payload_buf[0..h.len];
        try reader.readNoEof(payload);
        if (h.masked) unmask(payload, h.mask);
        switch (h.opcode) {
            .text, .binary => try writeFrame(writer, .text, payload),
            .ping => try writeFrame(writer, .pong, payload),
            .close => { try writeFrame(writer, .close, &[_]u8{}); break; },
            else => {},
        }
    }
    stream.close();
}

pub fn main() !void {
    const addr = try std.net.Address.parseIp4("127.0.0.1", 9001);
    var server = try addr.listen(.{ .reuse_address = true });
    std.debug.print("WebSocket server on ws://127.0.0.1:9001\n", .{});

    while (true) {
        const conn = try server.accept();
        _ = std.Thread.spawn(.{}, handleConnection, .{conn.stream}) catch |err| {
            std.log.err("spawn failed: {}", .{err});
            conn.stream.close();
        };
    }
}

继续阅读

探索更多技术文章

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

全部文章返回首页