WThinking

Appearance

Go技术专家进阶营（九）：WebSocket网关需求分析与架构设计 ​

上次上线一个实时推送功能，我用HTTP轮询凑合了一个月，结果服务器成本翻了三倍，老板看着账单问我"能不能换个方案"，我说能，于是开始研究WebSocket。这一研究就是两个月，踩的坑够写一本书。这篇就是那两个月的浓缩——从协议原理到架构设计，每一步都有血泪教训。

我是怕浪猫，一个在实时通信领域踩过无数坑的Go开发者。这是Go技术专家进阶营系列的第9篇，从这篇开始我们进入全新的模块——WebSocket网关。前8篇我们搞定了通知平台和权限系统，接下来几篇要攻克的是实时通信的核心基础设施。这篇先聊需求分析和架构设计，把地基打牢，后面写代码才不会心虚。

轮询不是实时通信，而是一种"假装实时"的妥协。真正的实时，是连接建立之后，数据随时可以双向流动。

在深入WebSocket之前，先说说为什么不用其他方案。HTTP轮询的痛点太明显了：客户端每隔3秒发一次请求，90%的请求返回空数据，服务器白白处理了大量无效请求。长轮询（Long Polling）稍微好一点，服务器hold住请求直到有数据才返回，但每次返回后又要重新建立连接，开销依然不小。SSE（Server-Sent Events）只能服务端到客户端单向推送，客户端到服务端还得走HTTP。

而WebSocket解决了所有这些问题：一次握手建立持久连接，双向实时通信，头部开销只有2-14字节（相比HTTP的几百字节头部），协议层面的心跳保活。这就是为什么几乎所有现代实时通信系统都选择WebSocket作为传输层。

选型不是选最新的，而是选最合适的。WebSocket不是银弹，但在实时通信场景下，它确实是目前最优解。

9.1 WebSocket协议深度解析 ​

搞WebSocket网关，第一步得把协议本身吃透。很多人觉得WebSocket就是"一个长连接"，这理解太浅了，浅到你在排查连接断开、消息丢失、跨域问题的时候会完全找不到方向。

9.1.1 从HTTP到WebSocket：协议升级机制 ​

WebSocket不是凭空出现的协议，它的握手阶段复用了HTTP协议。客户端先发一个HTTP请求，通过Upgrade头告诉服务器"我想升级到WebSocket协议"，服务器同意后返回101 Switching Protocols，之后这条TCP连接就不再走HTTP了，而是切换到WebSocket帧协议。

这个设计非常巧妙：复用HTTP协议的握手意味着WebSocket可以穿透大多数HTTP基础设施——反向代理、负载均衡器、CDN都能正确处理WebSocket升级请求。但这也带来了一些陷阱：某些中间代理可能会修改Upgrade头或者超时断开长时间闲置的连接，这些都需要在网关设计中考虑。比如Nginx默认的proxy_read_timeout是60秒，如果你的WebSocket心跳间隔大于60秒，Nginx就会把连接断掉。这个坑我在生产环境踩过，排查了整整一个下午。

来看一次完整的握手过程：

http
// 客户端请求
GET /ws/chat HTTP/1.1
Host: gateway.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Origin: https://example.com
Sec-WebSocket-Protocol: chat, superchat

// 服务端响应
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

这里有几个关键字段需要理解：

Sec-WebSocket-Key：客户端生成的随机Base64字符串，用来让服务端证明自己理解WebSocket协议。服务端把这个key拼接一个固定magic string 258EAFA5-E914-47DA-95CA-C5AB0DC85B11，做SHA-1哈希再Base64编码，得到Sec-WebSocket-Accept返回给客户端。这个过程不是安全机制，只是协议握手验证。

Sec-WebSocket-Version：协议版本号，当前标准是13。如果你看到客户端发的是其他版本，大概率是上古时代的浏览器或者某个不规范的SDK。

Sec-WebSocket-Protocol：子协议协商。客户端可以声明支持多个子协议，服务端选择一个返回。这个字段很重要，它允许你在WebSocket之上定义应用层协议，比如聊天协议、推送协议等。

Origin：客户端来源。服务端可以基于这个字段做跨域校验。注意WebSocket的跨域策略和HTTP CORS不一样，WebSocket没有预检请求，服务端需要自己校验Origin头。

理解协议握手不是学术练习，而是排查问题的第一道防线。连握手过程都不清楚，遇到连接失败你只能瞎猜。

用Go实现一个标准的握手响应：

go
package ws

import (
	"crypto/sha1"
	"encoding/base64"
	"fmt"
	"net/http"
	"strings"
)

const websocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

// HandshakeResponse 生成WebSocket握手响应
func HandshakeResponse(r *http.Request) (string, error) {
	key := r.Header.Get("Sec-WebSocket-Key")
	if key == "" {
		return "", fmt.Errorf("missing Sec-WebSocket-Key")
	}

	// 拼接magic string并计算SHA-1
	h := sha1.New()
	h.Write([]byte(key + websocketGUID))
	accept := base64.StdEncoding.EncodeToString(h.Sum(nil))

	return accept, nil
}

// UpgradeHandler WebSocket升级处理器
func UpgradeHandler(w http.ResponseWriter, r *http.Request) {
	// 校验方法
	if r.Method != http.MethodGet {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	// 校验Upgrade头
	if !strings.Contains(strings.ToLower(r.Header.Get("Connection")), "upgrade") {
		http.Error(w, "Invalid Connection header", http.StatusBadRequest)
		return
	}

	if strings.ToLower(r.Header.Get("Upgrade")) != "websocket" {
		http.Error(w, "Invalid Upgrade header", http.StatusBadRequest)
		return
	}

	// 校验版本
	if r.Header.Get("Sec-WebSocket-Version") != "13" {
		http.Error(w, "Unsupported WebSocket version", http.StatusBadRequest)
		return
	}

	// 生成握手响应
	accept, err := HandshakeResponse(r)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// 子协议协商
	protocols := r.Header.Values("Sec-WebSocket-Protocol")
	selectedProtocol := negotiateProtocol(protocols)

	// 写入响应头
	hijacker, ok := w.(http.Hijacker)
	if !ok {
		http.Error(w, "WebSocket upgrade not supported", http.StatusInternalServerError)
		return
	}

	conn, bufrw, err := hijacker.Hijack()
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	defer conn.Close()

	response := "HTTP/1.1 101 Switching Protocols\r\n" +
		"Upgrade: websocket\r\n" +
		"Connection: Upgrade\r\n" +
		"Sec-WebSocket-Accept: " + accept + "\r\n"

	if selectedProtocol != "" {
		response += "Sec-WebSocket-Protocol: " + selectedProtocol + "\r\n"
	}
	response += "\r\n"

	if _, err := bufrw.WriteString(response); err != nil {
		return
	}
	bufrw.Flush()

	// 此时TCP连接已升级为WebSocket，开始帧处理
	// 后续实现帧读取和写入...
}

func negotiateProtocol(protocols []string) string {
	supported := map[string]bool{
		"chat.v1":     true,
		"push.v1":     true,
		"presence.v1": true,
	}
	for _, p := range protocols {
		// 客户端可能发送 "chat.v1, push.v1" 形式
		for _, proto := range strings.Split(p, ",") {
			proto = strings.TrimSpace(proto)
			if supported[proto] {
				return proto
			}
		}
	}
	return ""
}

协议升级的本质是"借壳上市"——用HTTP的壳建立TCP连接，然后撕掉HTTP的皮，换成WebSocket的骨架。

9.1.2 WebSocket帧格式详解 ​

握手完成后，数据传输就完全走WebSocket帧格式了。理解帧格式是处理消息分片、压缩、控制帧的基础。

WebSocket帧格式如下：

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

关键字段解读：

FIN (1 bit)：是否是消息的最后一个分片。如果是分片消息，前面的分片FIN=0，最后一个分片FIN=1。

RSV1-3 (各1 bit)：保留位。如果协商了扩展（如permessage-deflate压缩），RSV1会被用于标记压缩。没有协商扩展时这些位必须为0，否则连接应被关闭。

opcode (4 bits)：帧类型。

  • 0x0： continuation frame（续帧，分片消息的后续分片）
  • 0x1： text frame（文本帧）
  • 0x2： binary frame（二进制帧）
  • 0x8： close frame（关闭帧）
  • 0x9： ping frame（心跳ping）
  • 0xA： pong frame（心跳pong）

MASK (1 bit)：客户端到服务端的消息必须mask，服务端到客户端的消息不能mask。这是为了防止中间代理缓存污染。

Payload length：7位、7+16位或7+64位三种编码方式。0-125直接用7位表示，126表示后2字节是长度，127表示后8字节是长度。

用Go实现一个帧解析器：

go
package ws

import (
	"encoding/binary"
	"fmt"
	"io"
)

// OpCode 帧类型
type OpCode byte

const (
	OpContinuation OpCode = 0x0
	OpText         OpCode = 0x1
	OpBinary       OpCode = 0x2
	OpClose        OpCode = 0x8
	OpPing         OpCode = 0x9
	OpPong         OpCode = 0xA
)

// Frame WebSocket帧
type Frame struct {
	FIN     bool
	OpCode  OpCode
	Masked  bool
	Payload []byte
}

// ReadFrame 从连接读取一个WebSocket帧
func ReadFrame(r io.Reader) (*Frame, error) {
	// 读取前2字节
	header := make([]byte, 2)
	if _, err := io.ReadFull(r, header); err != nil {
		return nil, fmt.Errorf("read frame header: %w", err)
	}

	fin := header[0]&0x80 != 0
	rsv1 := header[0]&0x40 != 0
	rsv2 := header[0]&0x20 != 0
	rsv3 := header[0]&0x10 != 0
	opcode := OpCode(header[0] & 0x0F)

	// 校验保留位
	if rsv1 || rsv2 || rsv3 {
		return nil, fmt.Errorf("reserved bits must be 0")
	}

	masked := header[1]&0x80 != 0
	payloadLen := int(header[1] & 0x7F)

	// 读取扩展长度
	switch payloadLen {
	case 126:
		ext := make([]byte, 2)
		if _, err := io.ReadFull(r, ext); err != nil {
			return nil, fmt.Errorf("read extended length: %w", err)
		}
		payloadLen = int(binary.BigEndian.Uint16(ext))
	case 127:
		ext := make([]byte, 8)
		if _, err := io.ReadFull(r, ext); err != nil {
			return nil, fmt.Errorf("read extended length: %w", err)
		}
		payloadLen = int(binary.BigEndian.Uint64(ext))
		// 校验最高位必须为0
		if ext[0]&0x80 != 0 {
			return nil, fmt.Errorf("invalid payload length")
		}
	}

	// 读取mask key
	var maskKey []byte
	if masked {
		maskKey = make([]byte, 4)
		if _, err := io.ReadFull(r, maskKey); err != nil {
			return nil, fmt.Errorf("read mask key: %w", err)
		}
	}

	// 读取payload
	payload := make([]byte, payloadLen)
	if _, err := io.ReadFull(r, payload); err != nil {
		return nil, fmt.Errorf("read payload: %w", err)
	}

	// 解除mask
	if masked {
		for i := range payload {
			payload[i] ^= maskKey[i%4]
		}
	}

	return &Frame{
		FIN:     fin,
		OpCode:  opcode,
		Masked:  masked,
		Payload: payload,
	}, nil
}

// WriteFrame 向连接写入一个WebSocket帧
func WriteFrame(w io.Writer, frame *Frame) error {
	var header [14]byte
	pos := 0

	// FIN + opcode
	if frame.FIN {
		header[0] |= 0x80
	}
	header[0] |= byte(frame.OpCode)

	// 服务端发送的帧不能mask
	maskBit := byte(0)
	if frame.Masked {
		maskBit = 0x80
	}

	// payload length
	payloadLen := len(frame.Payload)
	switch {
	case payloadLen <= 125:
		header[1] = maskBit | byte(payloadLen)
		pos = 2
	case payloadLen <= 65535:
		header[1] = maskBit | 126
		binary.BigEndian.PutUint16(header[2:], uint16(payloadLen))
		pos = 4
	default:
		header[1] = maskBit | 127
		binary.BigEndian.PutUint64(header[2:], uint64(payloadLen))
		pos = 10
	}

	// mask key
	if frame.Masked {
		maskKey := []byte{0x12, 0x34, 0x56, 0x78} // 实际应随机生成
		copy(header[pos:], maskKey)
		pos += 4

		// 写入header
		if _, err := w.Write(header[:pos]); err != nil {
			return err
		}

		// 写入masked payload
		masked := make([]byte, payloadLen)
		for i := range frame.Payload {
			masked[i] = frame.Payload[i] ^ maskKey[i%4]
		}
		_, err := w.Write(masked)
		return err
	}

	// 写入header + payload
	if _, err := w.Write(header[:pos]); err != nil {
		return err
	}
	_, err := w.Write(frame.Payload)
	return err
}

帧格式是WebSocket的底层语法，就像TCP段格式之于网络编程。不掌握帧格式，你连抓包分析都做不到。

9.1.3 消息分片机制 ​

WebSocket支持消息分片：一条逻辑消息可以被拆成多个帧发送。这在传输大消息时非常有用，比如发送一个10MB的文件，可以分成多个小帧，中间穿插控制帧（ping/pong）来维持连接活性。

分片规则：

  • 第一帧的opcode是Text(0x1)或Binary(0x2)，FIN=0
  • 后续帧的opcode是Continuation(0x0)，FIN=0
  • 最后一帧的opcode是Continuation(0x0)，FIN=1

实现一个消息读取器，处理分片消息：

go
package ws

import (
	"errors"
	"io"
)

// MessageReader 消息读取器，处理分片消息
type MessageReader struct {
	r io.Reader
}

// Message 读取一条完整的WebSocket消息
type Message struct {
	OpCode  OpCode
	Payload []byte
}

func (mr *MessageReader) ReadMessage() (*Message, error) {
	var fragments [][]byte
	var baseOpCode OpCode

	for {
		frame, err := ReadFrame(mr.r)
		if err != nil {
			return nil, err
		}

		switch frame.OpCode {
		case OpPing:
			// 收到ping，需要回复pong
			// 实际实现中需要异步回复
			continue
		case OpPong:
			// 收到pong，更新心跳时间
			continue
		case OpClose:
			return nil, errors.New("connection closed by peer")
		case OpText, OpBinary:
			// 新消息的第一帧
			if len(fragments) > 0 {
				return nil, errors.New("new message started before previous one completed")
			}
			baseOpCode = frame.OpCode
			fragments = append(fragments, frame.Payload)
		case OpContinuation:
			// 分片续帧
			if len(fragments) == 0 {
				return nil, errors.New("continuation frame without start frame")
			}
			fragments = append(fragments, frame.Payload)
		}

		if frame.FIN {
			// 消息完成，合并所有分片
			var totalLen int
			for _, f := range fragments {
				totalLen += len(f)
			}
			payload := make([]byte, 0, totalLen)
			for _, f := range fragments {
				payload = append(payload, f...)
			}
			return &Message{
				OpCode:  baseOpCode,
				Payload: payload,
			}, nil
		}
	}
}

分片是WebSocket处理大消息的标准机制。不分片，一条10MB的消息就能卡住整个连接的控制帧通道。

9.1.4 心跳机制与连接保活 ​

WebSocket连接是持久的TCP连接，但中间的代理、负载均衡器可能会因为长时间没有数据传输而断开连接。心跳机制就是定期发送Ping/Pong帧来保持连接活性。

WebSocket定义了两种心跳方式：

  • Ping (0x9)：一端发送Ping帧，另一端必须尽快回复Pong帧
  • Pong (0xA)：对Ping帧的响应

Ping/Pong帧可以携带最多125字节的payload，可以用来同步时间戳等信息。

实现一个心跳管理器：

go
package ws

import (
	"context"
	"log"
	"sync"
	"time"
)

// HeartbeatConfig 心跳配置
type HeartbeatConfig struct {
	// PingInterval 发送Ping的间隔
	PingInterval time.Duration
	// PongWait 等待Pong的超时时间
	PongWait time.Duration
	// WriteTimeout 写超时
	WriteTimeout time.Duration
}

// DefaultHeartbeatConfig 默认心跳配置
func DefaultHeartbeatConfig() *HeartbeatConfig {
	return &HeartbeatConfig{
		PingInterval: 30 * time.Second,
		PongWait:     10 * time.Second,
		WriteTimeout: 5 * time.Second,
	}
}

// HeartbeatManager 心跳管理器
type HeartbeatManager struct {
	config    *HeartbeatConfig
	conn      *Connection
	lastPong  time.Time
	mu        sync.RWMutex
	closeChan chan struct{}
}

// Connection 抽象连接接口
type Connection struct {
	writeChan chan *Frame
}

// NewHeartbeatManager 创建心跳管理器
func NewHeartbeatManager(conn *Connection, config *HeartbeatConfig) *HeartbeatManager {
	return &HeartbeatManager{
		config:    config,
		conn:      conn,
		lastPong:  time.Now(),
		closeChan: make(chan struct{}),
	}
}

// Start 启动心跳
func (h *HeartbeatManager) Start(ctx context.Context) {
	ticker := time.NewTicker(h.config.PingInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-h.closeChan:
			return
		case <-ticker.C:
			// 发送Ping帧
			pingFrame := &Frame{
				FIN:     true,
				OpCode:  OpPing,
				Payload: []byte(time.Now().Format(time.RFC3339Nano)),
			}

			select {
			case h.conn.writeChan <- pingFrame:
			default:
				log.Println("write channel full, skip ping")
			}

			// 检查Pong是否超时
			h.mu.RLock()
			lastPong := h.lastPong
			h.mu.RUnlock()

			if time.Since(lastPong) > h.config.PingInterval+h.config.PongWait {
				log.Println("pong timeout, closing connection")
				h.Close()
				return
			}
		}
	}
}

// HandlePong 处理收到的Pong帧
func (h *HeartbeatManager) HandlePong(payload []byte) {
	h.mu.Lock()
	h.lastPong = time.Now()
	h.mu.Unlock()

	// 可以从payload中解析时间戳，计算网络延迟
	if len(payload) > 0 {
		if t, err := time.Parse(time.RFC3339Nano, string(payload)); err == nil {
			rtt := time.Since(t)
			log.Printf("round trip time: %v", rtt)
		}
	}
}

// Close 关闭心跳
func (h *HeartbeatManager) Close() {
	select {
	case <-h.closeChan:
		// 已经关闭
	default:
		close(h.closeChan)
	}
}

心跳不是可选项，而是长连接的生命线。没有心跳的连接，就像不呼吸的人，撑不了多久。

9.1.5 连接关闭流程 ​

WebSocket的关闭是一个双向握手过程，比TCP的FIN更规范：

  1. 一端发送Close帧（opcode 0x8），可以携带关闭状态码和原因
  2. 另一端收到Close帧后，回复一个Close帧
  3. 底层TCP连接关闭

标准的关闭状态码：

状态码含义
1000正常关闭
1001端点离开（如页面关闭）
1002协议错误
1003不支持的数据类型
1006异常关闭（未发送Close帧）
1007数据格式错误
1008策略违规
1009消息过大
1011内部错误
4000-4999应用层自定义状态码
go
// CloseFrame 构建关闭帧
func CloseFrame(code uint16, reason string) *Frame {
	payload := make([]byte, 2+len(reason))
	binary.BigEndian.PutUint16(payload, code)
	copy(payload[2:], reason)
	return &Frame{
		FIN:     true,
		OpCode:  OpClose,
		Payload: payload,
	}
}

// ParseCloseFrame 解析关闭帧
func ParseCloseFrame(frame *Frame) (uint16, string, error) {
	if frame.OpCode != OpClose {
		return 0, "", errors.New("not a close frame")
	}
	if len(frame.Payload) == 0 {
		return 1005, "", nil // 1005表示没有状态码
	}
	if len(frame.Payload) < 2 {
		return 0, "", errors.New("invalid close frame payload")
	}
	code := binary.BigEndian.Uint16(frame.Payload[:2])
	reason := string(frame.Payload[2:])
	return code, reason, nil
}

关闭流程看似简单，却是很多连接泄漏问题的根源。不规范的关闭会导致服务端连接数持续增长，最终撑爆内存。

9.2 实时通信业务场景分析 ​

协议搞清楚了，接下来看业务。为什么要用WebSocket？什么场景需要WebSocket？这些问题的答案决定了你的架构设计方向。

9.2.1 场景全景图 ​

在我们的系统中，实时通信需求来自多个业务线。我把它们分为四类：

第一类：即时消息（IM）

典型场景是客服聊天、系统通知。消息量中等（每秒数百到数千条），但要求消息有序、不丢失、不重复。延迟要求在200ms以内。这类场景的核心挑战是消息的可靠投递和顺序保证。用户聊天消息如果乱序了，语义就完全变了——"我先说的yes再说的no"和"no然后yes"是两个完全不同的意思。

第二类：实时数据推送

典型场景是行情推送、监控告警。消息量大（每秒上万到十万条），允许少量丢消息但延迟要求极低（50ms以内）。这类场景通常不需要客户端到服务端的消息，是单向推送。行情推送中50ms的延迟差异可能导致交易策略的执行结果完全不同，所以这类场景对延迟的敏感度远高于IM。

第三类：协同编辑

典型场景是在线文档、白板。消息量不大但频率高，要求操作有序、冲突可解决。这类场景对延迟敏感，但对吞吐量要求不高。协同编辑的核心难点不在WebSocket传输层，而在上层的冲突解决算法（OT或CRDT）。但WebSocket层需要保证消息的顺序性，否则上层的冲突解决逻辑会变得异常复杂。

第四类：实时音视频信令

典型场景是WebRTC信令服务器。消息量小但实时性要求极高，连接需要快速建立和释放。信令消息通常只在连接建立阶段密集交换，一旦P2P连接建立，WebSocket信令通道就可以释放。这类场景对网关的要求是快速握手和低延迟转发。

第五类：在线状态同步

典型场景是显示用户在线/离线状态、输入中状态（typing indicator）。消息量不大但扇出广——一个人上线，可能需要通知他所有好友。这类场景的特点是高频小消息，对网关的消息聚合能力有要求。如果每个用户的好友列表平均100人，1万人同时上线就是100万条状态消息，这对网关的压力不亚于一次广播。

不同的业务场景对WebSocket网关的要求完全不同。用一套架构覆盖所有场景，要么过度设计，要么力不从心。

9.2.2 容量评估 ​

基于业务场景分析，我们对WebSocket网关的容量需求做了评估：

维度IM场景行情推送协同编辑综合
在线连接数5万10万1万16万
峰值消息QPS5千10万2千10.7万
单连接消息频率0.1条/秒1条/秒2条/秒-
消息平均大小2KB500B1KB-
出口带宽80Mbps400Mbps16Mbps500Mbps
延迟要求<200ms<50ms<100ms-

这个容量评估直接决定了我们的技术选型：

  • 单机需要支持至少2万连接（8台机器支撑16万连接，留30%冗余）
  • 消息转发需要亚毫秒级开销
  • 出口带宽需要考虑千兆网卡的限制（约900Mbps可用带宽）
  • 内存按每连接15KB估算，2万连接约300MB，加上其他开销，单机4GB内存足够
  • CPU主要消耗在帧解析和消息序列化，8核CPU可以满足2万连接的消息处理需求
  • Redis Cluster需要至少3主3从6个节点，每个节点4GB内存足够存储连接路由信息

容量评估不是拍脑袋，而是基于业务场景的数据推导。评估错了，要么资源浪费，要么线上炸了。每一个数字背后都应该有计算过程。

9.2.3 功能性需求清单 ​

基于业务场景，梳理出WebSocket网关的功能需求：

连接管理

  • 支持WebSocket协议握手与升级
  • 支持跨域校验（Origin白名单）
  • 支持子协议协商
  • 支持心跳保活（Ping/Pong）
  • 支持优雅关闭（Close帧交互）
  • 支持连接超时自动断开

消息路由

  • 支持点对点消息（私聊）
  • 支持群组消息（广播）
  • 支持房间消息（加入/离开房间）
  • 支持消息过滤与转换

会话管理

  • 支持用户多设备同时在线
  • 支持会话状态查询
  • 支持强制下线
  • 支持重连后会话恢复

可靠性保障

  • 支持消息ACK机制
  • 支持离线消息存储
  • 支持消息有序投递
  • 支持 Exactly-Once 语义（业务层去重）

运维监控

  • 连接数实时监控
  • 消息QPS监控
  • 慢连接检测
  • 异常断连告警

需求清单不是 Wish List，而是 Commitment List。列上去的每一项都要在后续设计中落实，否则就是给自己挖坑。

9.2.4 非功能性需求 ​

功能性需求决定"能不能做"，非功能性需求决定"做得好不好"。

性能指标

  • 单机连接数：≥ 2万
  • 消息转发延迟：P99 < 10ms（网关内部）
  • 消息吞吐量：≥ 5万 QPS（单机）
  • 内存占用：≤ 2GB（2万连接时）
  • CPU利用率：< 60%（峰值）

可用性指标

  • 网关可用性：99.95%
  • 连接成功率：> 99.9%
  • 异常断连率：< 0.1%/小时
  • 故障恢复时间：< 30秒

安全指标

  • 支持TLS加密传输（WSS）
  • 支持Token鉴权
  • 支持速率限制（单连接/单用户/全局）
  • 支持消息大小限制
  • 支持Origin白名单校验
  • 支持IP黑名单动态更新

可扩展性需求

  • 网关节点支持水平扩缩容，扩容时不停服
  • 连接注册中心支持分片扩展
  • 消息路由支持插件化扩展
  • 协议层支持子协议协商，方便业务迭代
  • 监控指标支持自定义扩展

9.2.5 技术约束与依赖 ​

在设计网关时，以下约束条件需要考虑：

基础设施约束

  • 部署在Kubernetes上，Pod的CPU/内存有Request/Limit限制
  • 服务发现依赖K8s Service或Consul
  • 配置中心使用Apollo或Nacos
  • 日志收集走ELK体系

团队能力约束

  • 团队Go经验丰富，但分布式系统经验有限
  • 运维团队对Redis熟悉，对Kafka一般
  • 前端团队对WebSocket了解不深，需要提供SDK

兼容性约束

  • 需要兼容老版本客户端的HTTP轮询降级
  • 需要支持微信小程序的WebSocket实现（有特殊限制）
  • 需要兼容iOS后台WebSocket断连恢复机制

9.3 WebSocket网关架构设计 ​

需求清楚了，开始设计架构。这部分我会把设计过程完整展现出来，包括被推翻的方案，因为"为什么不这样设计"和"该怎么设计"同样重要。

9.3.1 架构演进路径 ​

我没有一上来就设计一个完美架构，而是画了三个方案，逐一分析利弊，最终选了一个"不完美但务实"的方案。

方案一：单机WebSocket服务器

最简单的方案：一个Go进程，内置WebSocket服务器，直连Redis和消息队列。

优点：简单，开发快，适合MVP。 缺点：单点故障，无法水平扩展，连接数受单机限制。

单机方案不是技术选择，而是阶段选择。在验证阶段，单机方案是最优解；到了规模阶段，它就是最大的瓶颈。

方案二：多机WebSocket + Redis Pub/Sub

多个WebSocket节点通过Redis Pub/Sub同步消息。客户端连接任意节点，节点之间通过Redis转发消息。

优点：可以水平扩展连接数，开发复杂度适中。 缺点：Redis Pub/Sub不保证消息投递，节点故障时消息会丢失；连接数增大后Redis成为瓶颈。

方案三：WebSocket网关 + 消息总线

独立WebSocket网关层，后端服务通过消息总线（Kafka/RabbitMQ）与网关通信。网关只负责连接管理和消息转发，不处理业务逻辑。

优点：职责清晰，网关无状态可扩展，消息可靠性强。 缺点：架构复杂度高，链路长延迟增加，开发周期长。

最终我选了一个折中方案：方案二的增强版。核心思路是保留多机直连Redis的简洁性，但用Redis Cluster替代单节点Redis，并在网关层引入连接注册中心来解决跨节点消息路由问题。

方案三：WebSocket网关 + 消息总线

独立WebSocket网关层，后端服务通过消息总线（Kafka/RabbitMQ）与网关通信。网关只负责连接管理和消息转发，不处理业务逻辑。

优点：职责清晰，网关无状态可扩展，消息可靠性强。 缺点：架构复杂度高，链路长延迟增加，开发周期长。

方案四：基于Envoy/Istio的Sidecar模式

利用Service Mesh的WebSocket支持能力，每个Pod部署一个Sidecar代理WebSocket连接。业务进程只需要处理消息逻辑。

优点：基础设施层和应用层完全解耦，可观测性强。 缺点：Sidecar增加网络跳数和延迟，资源开销大，团队学习曲线陡峭。在我们当前的规模下属于杀鸡用牛刀。

架构设计不是选最美的方案，而是选当前阶段最合适的方案。过度设计和设计不足一样危险。

9.3.2 整体架构图 ​

                          +------------------+
                          |   Load Balancer  |
                          |   (HAProxy/LB)   |
                          +--------+---------+
                                   |
                 +-----------------+-----------------+
                 |                 |                 |
          +------+------+   +------+------+   +------+------+
          |  WS Node 1  |   |  WS Node 2  |   |  WS Node N  |
          | (Gateway)   |   | (Gateway)   |   | (Gateway)   |
          +------+------+   +------+------+   +------+------+
                 |                 |                 |
          +------+-----------------+-----------------+------+
          |                Connection Registry            |
          |                  (Redis Cluster)               |
          +------+-----------------+-----------------+------+
                 |                 |                 |
          +------+------+   +------+------+   +------+------+
          |  Redis Node1|   |  Redis Node2|   |  Redis Node3|
          |  (Slots 0-  |   |  (Slots     |   |  (Slots     |
          |   5460)     |   |  5461-10922)|   |  10923-16383)|
          +-------------+   +-------------+   +-------------+

每个WebSocket网关节点维护本地连接表，同时将连接信息注册到Redis Cluster。跨节点的消息路由通过Redis查询目标连接所在节点，然后通过节点间直接通信转发消息。

9.3.2.1 部署架构详解 ​

部署层面，WebSocket网关的架构有其特殊性。与普通HTTP服务不同，WebSocket是长连接，负载均衡不能简单地用Round Robin。一旦客户端连接到了某个网关节点，这条连接就会一直保持，直到客户端断开或服务端关闭。这意味着负载均衡的分配发生在连接建立时，而不是每次请求时。

负载均衡策略选择：

  • 最少连接数（Least Connections）：HAProxy的leastconn算法，新连接分配给当前连接数最少的节点。这是WebSocket网关最合适的策略，能自动平衡各节点的连接数。
  • 源地址哈希（Source IP Hash）：同一客户端IP始终哈希到同一节点。好处是节点故障时只有部分客户端受影响，坏处是可能导致负载不均。
  • 加权轮询（Weighted Round Robin）：根据节点配置的权重分配连接，适合节点配置不一致的场景。

我们选择最少连接数策略，配合健康检查。HAProxy每5秒检查一次网关节点的健康状态（通过HTTP接口/health），如果节点不健康则从负载均衡池中摘除。

Kubernetes部署时需要注意：WebSocket网关Pod不能随意重启，因为每次重启都会断开所有连接。需要配置terminationGracePeriodSeconds为足够长的时间（建议30秒以上），让优雅关闭流程有足够时间完成。同时使用Pod Disruption Budget确保滚动更新时至少保持一定数量的可用Pod。

负载均衡策略的选择直接影响连接分布的均匀程度。最少连接数策略虽然不是完美均衡，但在动态扩缩容场景下是最务实的选择。

9.3.3 网关节点架构 ​

单个网关节点的内部架构：

+-------------------------------------------------------------------+
|                     WebSocket Gateway Node                        |
|                                                                   |
|  +-------------+    +---------------+    +------------------+    |
|  |  Accepter   |--->|  Connection   |--->|  Read Loop       |    |
|  |  (TCP Listen)|   |  Manager      |    |  (Frame Reader)  |    |
|  +-------------+    +---------------+    +--------+---------+    |
|                                                   |              |
|  +-------------+    +---------------+             |              |
|  |  Write Loop |<---|  Message      |<------------+              |
|  | (Frame      |    |  Router       |                            |
|  |  Writer)    |    |               |                            |
|  +------+------+    +------+--------+                            |
|         |                  |                                      |
|         |          +-------+--------+    +------------------+    |
|         |          |  Auth Module   |    |  Heartbeat       |    |
|         |          |  (Token Verify)|    |  Manager         |    |
|         |          +----------------+    +------------------+    |
|         |                                                         |
|  +------+-----------+    +---------------+                       |
|  |  Registry Client  |    |  Metrics      |                       |
|  |  (Redis Register) |    |  Collector    |                       |
|  +-------------------+    +---------------+                       |
+-------------------------------------------------------------------+

各个模块的职责：

Accepter：监听TCP端口，接收新连接，完成WebSocket握手。它是网关的"前台接待"，负责第一印象。Accepter需要处理SYN flood攻击防护、连接速率限制、TLS握手加速等问题。在Go中，通常用一个独立的goroutine执行net.Listen和Accept循环，每接受一个连接就启动一个handler goroutine。

Connection Manager：管理所有在线连接的生命周期，包括创建、查找、关闭。它是网关的"户籍管理处"，每条连接的生老病死都归它管。设计Connection Manager时最关键的是并发安全——2万连接同时建立和断开时，连接表的锁不能成为瓶颈。

Read Loop：每条连接一个goroutine，循环读取WebSocket帧，解析后交给Message Router。Read Loop是最容易出goroutine泄漏的地方——如果连接断开但Read Loop没被正确终止，这个goroutine就会永远卡在io.ReadFull上。

Write Loop：每条连接一个goroutine，从发送队列读取消息并写入连接。Write Loop和Read Loop通过channel通信，实现了读写分离。这种设计避免了读写互斥锁，但需要注意send channel的背压处理——如果客户端消费太慢，channel满了该怎么办？

Message Router：根据消息类型和目标地址，路由消息到本地连接或远程节点。Router是网关的"快递分拣中心"，消息从四面八方来，要快速准确地送到正确的目的地。Router的性能直接决定了网关的消息转发延迟。

Auth Module：在握手阶段或握手后验证用户Token。Auth Module支持两种鉴权模式：握手阶段通过URL参数或Header传递Token（适合浏览器），握手后通过WebSocket消息发送Token（适合原生客户端）。Token验证通过后，用户信息会被关联到Connection对象上。

Heartbeat Manager：管理连接的心跳，检测死连接。除了标准Ping/Pong机制外，Heartbeat Manager还负责连接的TTL刷新——定期向注册中心续期，确保连接路由信息不过期。

Registry Client：将连接信息注册到Redis，支持跨节点查询。Registry Client封装了所有与Redis Cluster的交互，包括连接注册、注销、路由查询、TTL刷新等操作。它还需要处理Redis不可用时的降级逻辑。

Metrics Collector：采集运行指标，暴露给监控系统。关键指标包括当前连接数、消息QPS、消息延迟分布、连接建立/断开速率、Redis操作延迟等。Metrics Collector使用Prometheus格式暴露指标，配合Grafana做可视化。

9.3.5 WebSocket库选型 ​

Go生态中有几个主流的WebSocket库，选型时需要仔细对比：

gorilla/websocket：社区最知名的WebSocket库，功能完善，文档齐全。但已经进入维护模式，不再添加新功能。性能不错但不是最优。适合快速开发和对稳定性要求高的场景。

nhooyr/websocket（现更名为coder/websocket）：更现代的API设计，基于context的接口，零依赖。性能与gorilla相当，API更符合Go习惯。支持WebSocket压缩（permessage-deflate）。活跃维护中。

gobwas/ws：性能最高，零内存分配，但API比较底层，需要自己处理很多细节。适合对性能有极致要求的场景。

go-net/websocket：Go官方实验性实现，API简单但功能有限，不适合生产环境。

对比基准测试（在我的M1 Mac上用wrk压测）：

库连接建立QPS消息转发延迟(P99)内存/连接API易用性
gorilla/websocket8,2000.8ms12KB高
coder/websocket8,5000.7ms10KB高
gobwas/ws11,3000.4ms6KB低

最终我选择了coder/websocket。理由是：API设计现代且符合Go习惯，性能接近gobwas/ws，支持permessage-deflate压缩，活跃维护中。gobwas/ws虽然性能最优，但API过于底层，开发效率会受影响，而且我们单机2万连接的场景下，性能差异不显著。

库的选型不要只看benchmark数字，还要看维护活跃度、文档质量和团队学习成本。一个快10%但没人会用的库，比一个慢10%但上手即用的库更危险。

9.3.6 连接模型设计 ​

Go的goroutine模型非常适合WebSocket这种长连接场景。每条连接分配2个goroutine：一个读、一个写。通过channel在两者之间传递消息。

这个设计的内存开销需要仔细评估。一条连接的内存构成：goroutine初始栈2KB x 2 = 4KB，send channel缓冲区256个指针 = 2KB，Connection结构体约500B，net.Conn内部缓冲区约8KB，总计约15KB/连接。2万连接约300MB，完全在可控范围内。

但要注意Go的goroutine栈是动态增长的。如果消息处理逻辑中调用了深度递归或分配了大量局部变量，goroutine栈会从2KB增长到几MB。2万连接的goroutine如果都增长到8KB，内存就从300MB变成了1.2GB。所以消息处理要尽量用异步模式，避免在read goroutine中执行重逻辑。

每条连接两个goroutine的设计是Go网络编程的经典模式。简洁、高效、易于理解。但2万连接意味着4万goroutine，你得确保goroutine的内存开销在可控范围内。

go
package gateway

import (
	"context"
	"log"
	"net"
	"sync"
	"time"

	"example.com/ws"
)

// Connection WebSocket连接封装
type Connection struct {
	ID        string
	UserID    string
	DeviceID  string
	conn      net.Conn
	sendChan  chan *ws.Frame
	closeChan chan struct{}
	once      sync.Once
	createAt  time.Time
	lastActive time.Time
	mu        sync.RWMutex
}

// ConnectionConfig 连接配置
type ConnectionConfig struct {
	SendBufferSize    int
	ReadTimeout       time.Duration
	WriteTimeout      time.Duration
	MaxMessageSize    int64
}

// DefaultConnectionConfig 默认连接配置
func DefaultConnectionConfig() *ConnectionConfig {
	return &ConnectionConfig{
		SendBufferSize:  256,
		ReadTimeout:     60 * time.Second,
		WriteTimeout:    10 * time.Second,
		MaxMessageSize:  1 << 20, // 1MB
	}
}

// NewConnection 创建新连接
func NewConnection(id, userID, deviceID string, conn net.Conn, config *ConnectionConfig) *Connection {
	return &Connection{
		ID:        id,
		UserID:    userID,
		DeviceID:  deviceID,
		conn:      conn,
		sendChan:  make(chan *ws.Frame, config.SendBufferSize),
		closeChan: make(chan struct{}),
		createAt:  time.Now(),
		lastActive: time.Now(),
	}
}

// StartReadLoop 启动读取循环
func (c *Connection) StartReadLoop(ctx context.Context, handler MessageHandler) {
	defer c.Close()

	reader := &ws.MessageReader{R: c.conn}

	for {
		select {
		case <-ctx.Done():
			return
		case <-c.closeChan:
			return
		default:
		}

		msg, err := reader.ReadMessage()
		if err != nil {
			if isClosedError(err) {
				log.Printf("connection %s closed", c.ID)
			} else {
				log.Printf("connection %s read error: %v", c.ID, err)
			}
			return
		}

		c.mu.Lock()
		c.lastActive = time.Now()
		c.mu.Unlock()

		// 异步处理消息
		go handler.Handle(ctx, c, msg)
	}
}

// StartWriteLoop 启动写入循环
func (c *Connection) StartWriteLoop(ctx context.Context) {
	defer c.Close()

	for {
		select {
		case <-ctx.Done():
			return
		case <-c.closeChan:
			return
		case frame := <-c.sendChan:
			c.conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
			if err := ws.WriteFrame(c.conn, frame); err != nil {
				log.Printf("connection %s write error: %v", c.ID, err)
				return
			}
		}
	}
}

// SendMessage 发送消息到连接
func (c *Connection) SendMessage(payload []byte, opcode ws.OpCode) error {
	frame := &ws.Frame{
		FIN:     true,
		OpCode:  opcode,
		Payload: payload,
	}
	select {
	case c.sendChan <- frame:
		return nil
	default:
		// 发送队列满了，说明客户端消费太慢
		return fmt.Errorf("send buffer full, connection %s", c.ID)
	}
}

// Close 关闭连接
func (c *Connection) Close() {
	c.once.Do(func() {
		close(c.closeChan)
		c.conn.Close()
	})
}

// IsAlive 检查连接是否存活
func (c *Connection) IsAlive() bool {
	select {
	case <-c.closeChan:
		return false
	default:
		return true
	}
}

// GetLastActive 获取最后活跃时间
func (c *Connection) GetLastActive() time.Time {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.lastActive
}

func isClosedError(err error) bool {
	return err.Error() == "connection closed by peer" || 
		err.Error() == "EOF"
}

每条连接两个goroutine的设计是Go网络编程的经典模式。简洁、高效、易于理解。但2万连接意味着4万goroutine，你得确保goroutine的内存开销在可控范围内。

9.3.5 消息路由设计 ​

消息路由是网关的核心功能。一条消息从客户端到达网关后，需要确定目标接收者并将其转发到正确的连接。

消息路由分为三种模式：

点对点路由：消息发送给指定用户。需要查询用户当前连接在哪个节点上。

群组路由：消息发送给群组所有成员。需要获取群组成员列表，然后逐一或批量发送。

广播路由：消息发送给所有在线连接。直接遍历本地连接表发送。

go
package gateway

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"example.com/ws"
)

// MessageHandler 消息处理器接口
type MessageHandler interface {
	Handle(ctx context.Context, conn *Connection, msg *ws.Message)
}

// Router 消息路由器
type Router struct {
	localConns   *ConnectionTable
	registry     *RegistryClient
	rpcClient    *NodeRPCClient
	routeTimeout time.Duration
}

// NewRouter 创建路由器
func NewRouter(localConns *ConnectionTable, registry *RegistryClient, rpcClient *NodeRPCClient) *Router {
	return &Router{
		localConns:   localConns,
		registry:     registry,
		rpcClient:    rpcClient,
		routeTimeout: 3 * time.Second,
	}
}

// RouteMessage 路由消息
func (r *Router) RouteMessage(ctx context.Context, msg *OutboundMessage) error {
	ctx, cancel := context.WithTimeout(ctx, r.routeTimeout)
	defer cancel()

	switch msg.RouteType {
	case RouteTypePointToPoint:
		return r.routePointToPoint(ctx, msg)
	case RouteTypeGroup:
		return r.routeGroup(ctx, msg)
	case RouteTypeBroadcast:
		return r.routeBroadcast(ctx, msg)
	default:
		return fmt.Errorf("unknown route type: %d", msg.RouteType)
	}

}

// OutboundMessage 出站消息
type OutboundMessage struct {
	RouteType  RouteType
	TargetID   string   // 用户ID(点对点)或群组ID(群组)
	ExcludeID  string   // 排除的连接ID(发送者)
	Payload    []byte
	OpCode     ws.OpCode
}

// RouteType 路由类型
type RouteType int

const (
	RouteTypePointToPoint RouteType = iota
	RouteTypeGroup
	RouteTypeBroadcast
)

// routePointToPoint 点对点路由
func (r *Router) routePointToPoint(ctx context.Context, msg *OutboundMessage) error {
	// 1. 查找用户连接所在节点
	connInfos, err := r.registry.GetUserConnections(ctx, msg.TargetID)
	if err != nil {
		return fmt.Errorf("query user connections: %w", err)
	}

	if len(connInfos) == 0 {
		// 用户不在线，可以走离线消息存储
		return ErrUserOffline
	}

	// 2. 逐个连接转发
	for _, info := range connInfos {
		if info.NodeID == r.localNodeID {
			// 本地连接，直接发送
			conn := r.localConns.Get(info.ConnID)
			if conn != nil && conn.IsAlive() {
				if err := conn.SendMessage(msg.Payload, msg.OpCode); err != nil {
					log.Printf("send to local conn %s failed: %v", info.ConnID, err)
				}
			}
		} else {
			// 远程连接，通过RPC转发
			if err := r.rpcClient.ForwardMessage(ctx, info.NodeID, info.ConnID, msg.Payload, msg.OpCode); err != nil {
				log.Printf("forward to node %s conn %s failed: %v", info.NodeID, info.ConnID, err)
			}
		}
	}

	return nil
}

// routeGroup 群组路由
func (r *Router) routeGroup(ctx context.Context, msg *OutboundMessage) error {
	// 1. 获取群组成员列表
	members, err := r.registry.GetGroupMembers(ctx, msg.TargetID)
	if err != nil {
		return fmt.Errorf("query group members: %w", err)
	}

	// 2. 按节点分组，减少RPC调用次数
	nodeBatch := make(map[string][]string) // nodeID -> []connID
	for _, member := range members {
		if member.UserID == msg.ExcludeID {
			continue
		}
		connInfos, _ := r.registry.GetUserConnections(ctx, member.UserID)
		for _, info := range connInfos {
			if info.NodeID == r.localNodeID {
				// 本地连接直接发送
				conn := r.localConns.Get(info.ConnID)
				if conn != nil && conn.IsAlive() {
					conn.SendMessage(msg.Payload, msg.OpCode)
				}
			} else {
				nodeBatch[info.NodeID] = append(nodeBatch[info.NodeID], info.ConnID)
			}
		}
	}

	// 3. 批量转发到远程节点
	for nodeID, connIDs := range nodeBatch {
		if err := r.rpcClient.BatchForward(ctx, nodeID, connIDs, msg.Payload, msg.OpCode); err != nil {
			log.Printf("batch forward to node %s failed: %v", nodeID, err)
		}
	}

	return nil
}

// routeBroadcast 广播路由
func (r *Router) routeBroadcast(ctx context.Context, msg *OutboundMessage) error {
	// 1. 广播到本地所有连接
	r.localConns.Foreach(func(conn *Connection) {
		if conn.ID != msg.ExcludeID && conn.IsAlive() {
			conn.SendMessage(msg.Payload, msg.OpCode)
		}
	})

	// 2. 广播到所有其他节点
	nodes, err := r.registry.GetAllNodes(ctx)
	if err != nil {
		return fmt.Errorf("query all nodes: %w", err)
	}

	var wg sync.WaitGroup
	for _, nodeID := range nodes {
		if nodeID == r.localNodeID {
			continue
		}
		wg.Add(1)
		go func(nid string) {
			defer wg.Done()
			if err := r.rpcClient.Broadcast(ctx, nid, msg.ExcludeID, msg.Payload, msg.OpCode); err != nil {
				log.Printf("broadcast to node %s failed: %v", nid, err)
			}
		}(nodeID)
	}
	wg.Wait()

	return nil
}

路由设计的关键不是怎么发消息，而是怎么找到消息该去哪。路由效率决定了网关的延迟下限。

9.4 连接管理与会话保持方案 ​

连接管理是WebSocket网关最核心的模块。2万条连接的创建、查找、关闭，看似简单，但当你真正面对并发、内存、GC压力的时候，每个细节都是坑。

9.4.1 连接表设计 ​

连接表是网关的核心数据结构，需要支持以下操作：

  • 按连接ID查找连接
  • 按用户ID查找该用户的所有连接
  • 添加新连接
  • 删除连接
  • 遍历所有连接（广播场景）

我用了双索引结构：一个用连接ID做key，一个用用户ID做key。

go
package gateway

import (
	"sync"
	"time"
)

// ConnectionTable 连接表
type ConnectionTable struct {
	mu            sync.RWMutex
	byConnID      map[string]*Connection
	byUserID      map[string]map[string]*Connection // userID -> connID -> conn
	maxConns      int
	currentConns  int
}

// NewConnectionTable 创建连接表
func NewConnectionTable(maxConns int) *ConnectionTable {
	return &ConnectionTable{
		byConnID: make(map[string]*Connection),
		byUserID: make(map[string]map[string]*Connection),
		maxConns: maxConns,
	}
}

// Add 添加连接
func (t *ConnectionTable) Add(conn *Connection) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	if t.currentConns >= t.maxConns {
		return ErrMaxConnectionsExceeded
	}

	// 按连接ID索引
	t.byConnID[conn.ID] = conn

	// 按用户ID索引
	if t.byUserID[conn.UserID] == nil {
		t.byUserID[conn.UserID] = make(map[string]*Connection)
	}
	t.byUserID[conn.UserID][conn.ID] = conn

	t.currentConns++
	return nil
}

// Get 按连接ID查找
func (t *ConnectionTable) Get(connID string) *Connection {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.byConnID[connID]
}

// GetByUserID 按用户ID查找所有连接
func (t *ConnectionTable) GetByUserID(userID string) []*Connection {
	t.mu.RLock()
	defer t.mu.RUnlock()

	conns := make([]*Connection, 0, len(t.byUserID[userID]))
	for _, conn := range t.byUserID[userID] {
		conns = append(conns, conn)
	}
	return conns
}

// Remove 删除连接
func (t *ConnectionTable) Remove(connID string) {
	t.mu.Lock()
	defer t.mu.Unlock()

	conn, ok := t.byConnID[connID]
	if !ok {
		return
	}

	delete(t.byConnID, connID)

	if userConns, ok := t.byUserID[conn.UserID]; ok {
		delete(userConns, connID)
		if len(userConns) == 0 {
			delete(t.byUserID, conn.UserID)
		}
	}

	t.currentConns--
}

// Foreach 遍历所有连接
func (t *ConnectionTable) Foreach(fn func(*Connection)) {
	t.mu.RLock()
	defer t.mu.RUnlock()

	for _, conn := range t.byConnID {
		fn(conn)
	}
}

// Count 返回当前连接数
func (t *ConnectionTable) Count() int {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.currentConns
}

// CleanInactive 清理不活跃的连接
func (t *ConnectionTable) CleanInactive(timeout time.Duration) int {
	t.mu.Lock()
	defer t.mu.Unlock()

	now := time.Now()
	cleaned := 0

	for connID, conn := range t.byConnID {
		if now.Sub(conn.GetLastActive()) > timeout {
			conn.Close()
			delete(t.byConnID, connID)

			if userConns, ok := t.byUserID[conn.UserID]; ok {
				delete(userConns, connID)
				if len(userConns) == 0 {
					delete(t.byUserID, conn.UserID)
				}
			}

			t.currentConns--
			cleaned++
		}
	}

	return cleaned
}

数据结构的选择看似是细节，实则是架构的基石。双索引结构用空间换时间，在查找密集的场景下是值得的。

9.4.2 连接注册中心 ​

单节点的连接表只能管理本节点的连接。跨节点的消息路由需要一个分布式的连接注册中心，记录每个用户当前连接在哪个节点上。

我选择Redis Cluster作为注册中心，原因有三：

  1. 读写延迟低（亚毫秒级），适合连接注册这种高频操作
  2. Redis Cluster提供水平扩展能力，不会成为单点瓶颈
  3. TTL机制天然支持连接超时清理
go
package gateway

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// ConnInfo 连接信息
type ConnInfo struct {
	ConnID   string    `json:"conn_id"`
	NodeID   string    `json:"node_id"`
	UserID   string    `json:"user_id"`
	DeviceID string    `json:"device_id"`
	CreateAt time.Time `json:"create_at"`
}

// RegistryClient 连接注册中心客户端
type RegistryClient struct {
	rdb         *redis.ClusterClient
	nodeID      string
	keyPrefix   string
	keyTTL      time.Duration
}

// NewRegistryClient 创建注册中心客户端
func NewRegistryClient(rdb *redis.ClusterClient, nodeID string) *RegistryClient {
	return &RegistryClient{
		rdb:       rdb,
		nodeID:    nodeID,
		keyPrefix: "ws:gateway:",
		keyTTL:    90 * time.Second, // 连接TTL，需要大于心跳间隔
	}
}

// RegisterConnection 注册连接
func (r *RegistryClient) RegisterConnection(ctx context.Context, info *ConnInfo) error {
	// 连接信息存入两个key：
	// 1. ws:gateway:conn:{connID} -> 连接详情
	// 2. ws:gateway:user:{userID} -> Set of connID

	data, err := json.Marshal(info)
	if err != nil {
		return fmt.Errorf("marshal conn info: %w", err)
	}

	connKey := fmt.Sprintf("%sconn:%s", r.keyPrefix, info.ConnID)
	userKey := fmt.Sprintf("%suser:%s", r.keyPrefix, info.UserID)

	pipe := r.rdb.Pipeline()
	pipe.Set(ctx, connKey, data, r.keyTTL)
	pipe.SAdd(ctx, userKey, info.ConnID)
	pipe.Expire(ctx, userKey, r.keyTTL)

	// 同时注册到节点集合，用于广播
	nodeKey := fmt.Sprintf("%snode:%s", r.keyPrefix, r.nodeID)
	pipe.SAdd(ctx, nodeKey, info.ConnID)
	pipe.Expire(ctx, nodeKey, r.keyTTL)

	_, err = pipe.Exec(ctx)
	return err
}

// UnregisterConnection 注销连接
func (r *RegistryClient) UnregisterConnection(ctx context.Context, connID, userID string) error {
	connKey := fmt.Sprintf("%sconn:%s", r.keyPrefix, connID)
	userKey := fmt.Sprintf("%suser:%s", r.keyPrefix, userID)
	nodeKey := fmt.Sprintf("%snode:%s", r.keyPrefix, r.nodeID)

	pipe := r.rdb.Pipeline()
	pipe.Del(ctx, connKey)
	pipe.SRem(ctx, userKey, connID)
	pipe.SRem(ctx, nodeKey, connID)

	_, err := pipe.Exec(ctx)
	return err
}

// GetUserConnections 获取用户的所有连接信息
func (r *RegistryClient) GetUserConnections(ctx context.Context, userID string) ([]*ConnInfo, error) {
	userKey := fmt.Sprintf("%suser:%s", r.keyPrefix, userID)

	// 获取用户所有连接ID
	connIDs, err := r.rdb.SMembers(ctx, userKey).Result()
	if err != nil {
		return nil, fmt.Errorf("get user conn ids: %w", err)
	}

	if len(connIDs) == 0 {
		return nil, nil
	}

	// 批量获取连接详情
	keys := make([]string, 0, len(connIDs))
	for _, connID := range connIDs {
		keys = append(keys, fmt.Sprintf("%sconn:%s", r.keyPrefix, connID))
	}

	results, err := r.rdb.MGet(ctx, keys...).Result()
	if err != nil {
		return nil, fmt.Errorf("mget conn info: %w", err)
	}

	conns := make([]*ConnInfo, 0, len(results))
	for i, result := range results {
		if result == nil {
			// 连接信息已过期，清理脏数据
			r.rdb.SRem(ctx, userKey, connIDs[i])
			continue
		}

		data, ok := result.(string)
		if !ok {
			continue
		}

		info := &ConnInfo{}
		if err := json.Unmarshal([]byte(data), info); err != nil {
			continue
		}
		conns = append(conns, info)
	}

	return conns, nil
}

// GetAllNodes 获取所有在线节点
func (r *RegistryClient) GetAllNodes(ctx context.Context) ([]string, error) {
	// 使用一个全局Set记录所有节点
	nodeSetKey := fmt.Sprintf("%snodes", r.keyPrefix)
	nodes, err := r.rdb.SMembers(ctx, nodeSetKey).Result()
	if err != nil {
		return nil, fmt.Errorf("get all nodes: %w", err)
	}

	// 过滤掉过期的节点
	var aliveNodes []string
	for _, nodeID := range nodes {
		nodeKey := fmt.Sprintf("%snode:%s", r.keyPrefix, nodeID)
		exists, err := r.rdb.Exists(ctx, nodeKey).Result()
		if err != nil {
			continue
		}
		if exists > 0 {
			aliveNodes = append(aliveNodes, nodeID)
		} else {
			// 节点已下线，清理
			r.rdb.SRem(ctx, nodeSetKey, nodeID)
		}
	}

	return aliveNodes, nil
}

// RefreshTTL 刷新连接TTL（心跳时调用）
func (r *RegistryClient) RefreshTTL(ctx context.Context, connID, userID string) error {
	connKey := fmt.Sprintf("%sconn:%s", r.keyPrefix, connID)
	userKey := fmt.Sprintf("%suser:%s", r.keyPrefix, userID)
	nodeKey := fmt.Sprintf("%snode:%s", r.keyPrefix, r.nodeID)

	pipe := r.rdb.Pipeline()
	pipe.Expire(ctx, connKey, r.keyTTL)
	pipe.Expire(ctx, userKey, r.keyTTL)
	pipe.Expire(ctx, nodeKey, r.keyTTL)

	_, err := pipe.Exec(ctx)
	return err
}

// RegisterNode 注册节点
func (r *RegistryClient) RegisterNode(ctx context.Context) error {
	nodeSetKey := fmt.Sprintf("%snodes", r.keyPrefix)
	nodeKey := fmt.Sprintf("%snode:%s", r.keyPrefix, r.nodeID)

	pipe := r.rdb.Pipeline()
	pipe.SAdd(ctx, nodeSetKey, r.nodeID)
	// 用一个虚拟key标记节点存活
	pipe.Set(ctx, nodeKey, "alive", r.keyTTL)

	_, err := pipe.Exec(ctx)
	return err
}

注册中心是分布式网关的"电话簿"。没有它，你连消息该往哪个节点发都不知道。

9.4.3 会话恢复机制 ​

客户端网络波动导致的断连重连是WebSocket网关必须处理的场景。如果每次重连都要重新建立会话上下文，用户体验会很差。

会话恢复的核心思路：

  1. 连接断开时，保留会话上下文一段缓冲时间
  2. 客户端重连时携带之前的会话ID
  3. 网关找到缓存的会话上下文，恢复会话
  4. 投递断连期间积压的消息
go
package gateway

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

// Session 会话上下文
type Session struct {
	SessionID   string                 `json:"session_id"`
	UserID      string                 `json:"user_id"`
	DeviceID    string                 `json:"device_id"`
	JoinedRooms []string               `json:"joined_rooms"`
	LastSeqNum  int64                  `json:"last_seq_num"`
	CustomData  map[string]interface{} `json:"custom_data,omitempty"`
}

// SessionManager 会话管理器
type SessionManager struct {
	rdb        *redis.ClusterClient
	keyPrefix  string
	bufferTime time.Duration
}

// NewSessionManager 创建会话管理器
func NewSessionManager(rdb *redis.ClusterClient) *SessionManager {
	return &SessionManager{
		rdb:        rdb,
		keyPrefix:  "ws:session:",
		bufferTime: 5 * time.Minute, // 会话缓冲时间
	}
}

// SaveSession 保存会话（连接断开时调用）
func (m *SessionManager) SaveSession(ctx context.Context, session *Session) error {
	data, err := json.Marshal(session)
	if err != nil {
		return fmt.Errorf("marshal session: %w", err)
	}

	key := fmt.Sprintf("%s%s", m.keyPrefix, session.SessionID)
	return m.rdb.Set(ctx, key, data, m.bufferTime).Err()
}

// RestoreSession 恢复会话（重连时调用）
func (m *SessionManager) RestoreSession(ctx context.Context, sessionID string) (*Session, error) {
	key := fmt.Sprintf("%s%s", m.keyPrefix, sessionID)

	data, err := m.rdb.Get(ctx, key).Bytes()
	if err == redis.Nil {
		return nil, ErrSessionNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("get session: %w", err)
	}

	session := &Session{}
	if err := json.Unmarshal(data, session); err != nil {
		return nil, fmt.Errorf("unmarshal session: %w", err)
	}

	// 恢复后删除缓冲的会话
	m.rdb.Del(ctx, key)

	return session, nil
}

// DeleteSession 删除会话
func (m *SessionManager) DeleteSession(ctx context.Context, sessionID string) error {
	key := fmt.Sprintf("%s%s", m.keyPrefix, sessionID)
	return m.rdb.Del(ctx, key).Err()
}

// MessageBuffer 消息缓冲（断连期间的消息）
type MessageBuffer struct {
	rdb        *redis.ClusterClient
	keyPrefix  string
	maxBuffer  int64
	bufferTime time.Duration
}

// NewMessageBuffer 创建消息缓冲
func NewMessageBuffer(rdb *redis.ClusterClient) *MessageBuffer {
	return &MessageBuffer{
		rdb:        rdb,
		keyPrefix:  "ws:buffer:",
		maxBuffer:  100,             // 最多缓冲100条消息
		bufferTime: 5 * time.Minute, // 缓冲5分钟
	}
}

// PushMessage 推入缓冲消息
func (b *MessageBuffer) PushMessage(ctx context.Context, userID string, msg []byte) error {
	key := fmt.Sprintf("%s%s", b.keyPrefix, userID)

	pipe := b.rdb.Pipeline()
	pipe.LPush(ctx, key, msg)
	pipe.LTrim(ctx, key, 0, b.maxBuffer-1) // 保留最新的N条
	pipe.Expire(ctx, key, b.bufferTime)

	_, err := pipe.Exec(ctx)
	return err
}

// PopMessages 取出所有缓冲消息
func (b *MessageBuffer) PopMessages(ctx context.Context, userID string) ([][]byte, error) {
	key := fmt.Sprintf("%s%s", b.keyPrefix, userID)

	// 取出所有消息并删除key
	msgs, err := b.rdb.LRange(ctx, key, 0, -1).Result()
	if err != nil {
		return nil, err
	}

	b.rdb.Del(ctx, key)

	result := make([][]byte, 0, len(msgs))
	// LPush是头插法，LRange取出来是倒序的，需要反转
	for i := len(msgs) - 1; i >= 0; i-- {
		result = append(result, []byte(msgs[i]))
	}

	return result, nil
}

会话恢复不是可选项，而是实时通信系统的标配。用户网络切换那几秒钟的消息，不能就这么丢了。

9.4.4 多设备在线管理 ​

同一个用户可能同时有手机、平板、电脑多个设备在线。WebSocket网关需要正确处理多设备场景：

  1. 同一用户的多条连接独立管理，各有自己的连接ID
  2. 点对点消息需要投递到该用户的所有在线设备
  3. 支持设备级别的消息排除（A设备发的消息不回传给A设备）
  4. 支持设备优先级（某些消息只投递到主设备）
go
// DeviceType 设备类型
type DeviceType string

const (
	DeviceMobile  DeviceType = "mobile"
	DeviceTablet  DeviceType = "tablet"
	DeviceDesktop DeviceType = "desktop"
	DeviceWeb     DeviceType = "web"
)

// DeviceConnection 设备连接信息
type DeviceConnection struct {
	ConnID     string     `json:"conn_id"`
	DeviceType DeviceType `json:"device_type"`
	DeviceID   string     `json:"device_id"`
	IsPrimary  bool       `json:"is_primary"`
}

// MultiDeviceManager 多设备管理器
type MultiDeviceManager struct {
	localConns *ConnectionTable
	registry   *RegistryClient
}

// SendToUser 发送消息到用户的所有设备
func (m *MultiDeviceManager) SendToUser(ctx context.Context, userID string, payload []byte, opcode ws.OpCode, excludeConnID string) error {
	// 获取用户所有连接
	connInfos, err := m.registry.GetUserConnections(ctx, userID)
	if err != nil {
		return err
	}

	for _, info := range connInfos {
		if info.ConnID == excludeConnID {
			continue
		}

		// 尝试本地发送
		if info.NodeID == m.localNodeID {
			conn := m.localConns.Get(info.ConnID)
			if conn != nil && conn.IsAlive() {
				conn.SendMessage(payload, opcode)
			}
		}
		// 远程节点通过RPC转发，略...
	}

	return nil
}

// SendToPrimaryDevice 只发送到主设备
func (m *MultiDeviceManager) SendToPrimaryDevice(ctx context.Context, userID string, payload []byte, opcode ws.OpCode) error {
	connInfos, err := m.registry.GetUserConnections(ctx, userID)
	if err != nil {
		return err
	}

	// 查找主设备连接
	for _, info := range connInfos {
		// 主设备判断逻辑：优先Desktop > Tablet > Mobile
		// 实际应根据业务场景设计
		if info.NodeID == m.localNodeID {
			conn := m.localConns.Get(info.ConnID)
			if conn != nil && conn.IsAlive() {
				return conn.SendMessage(payload, opcode)
			}
		}
	}

	return ErrNoActiveDevice
}

多设备支持看起来是"多发一份消息"的事，实际上涉及设备优先级、消息去重、状态同步等一整套机制。

9.4.5 优雅关闭与连接清理 ​

网关在发布新版本或缩容时，需要优雅地关闭连接。粗暴地断开TCP连接会导致客户端消息丢失和重连风暴。

优雅关闭流程：

  1. 停止接收新连接（从负载均衡摘除）
  2. 给所有在线连接发送Close帧（状态码1001，表示服务端离开）
  3. 等待客户端主动断开（最多等待10秒）
  4. 超时后强制关闭残留连接
  5. 从注册中心注销节点
  6. 关闭进程
go
package gateway

import (
	"context"
	"log"
	"sync"
	"time"

	"example.com/ws"
)

// GracefulShutdown 优雅关闭
type GracefulShutdown struct {
	connTable   *ConnectionTable
	registry    *RegistryClient
	heartbeat   *HeartbeatManager
	waitTimeout time.Duration
}

// NewGracefulShutdown 创建优雅关闭器
func NewGracefulShutdown(connTable *ConnectionTable, registry *RegistryClient) *GracefulShutdown {
	return &GracefulShutdown{
		connTable:   connTable,
		registry:    registry,
		waitTimeout: 10 * time.Second,
	}
}

// Shutdown 执行优雅关闭
func (g *GracefulShutdown) Shutdown(ctx context.Context) error {
	log.Println("starting graceful shutdown...")

	// 1. 停止心跳，不再维持连接
	// g.heartbeat.Stop()

	// 2. 给所有连接发送Close帧
	closeFrame := ws.CloseFrame(1001, "server shutting down")

	var wg sync.WaitGroup
	g.connTable.Foreach(func(conn *Connection) {
		wg.Add(1)
		go func(c *Connection) {
			defer wg.Done()
			// 发送Close帧
			c.SendMessage(closeFrame.Payload, ws.OpClose)
		}(conn)
	})

	// 3. 等待连接关闭或超时
	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()

	select {
	case <-done:
		log.Println("all connections closed gracefully")
	case <-time.After(g.waitTimeout):
		log.Printf("timeout waiting for connections to close, force closing %d connections",
			g.connTable.Count())
		// 强制关闭所有连接
		g.connTable.Foreach(func(conn *Connection) {
			conn.Close()
		})
	}

	// 4. 从注册中心注销节点
	if err := g.registry.UnregisterNode(ctx); err != nil {
		log.Printf("unregister node failed: %v", err)
	}

	log.Println("graceful shutdown complete")
	return nil
}

优雅关闭体现的是对用户的尊重。你给客户端一个"我要走了"的信号，比直接消失要体面得多。

9.4.6 安全设计 ​

WebSocket网关作为直接暴露在公网的服务，安全设计不容忽视。以下是几个关键的安全措施：

TLS终止与证书管理

WebSocket的ws://协议是明文传输，生产环境必须使用wss://（WebSocket over TLS）。TLS终止可以放在两个位置：负载均衡层或网关层。

放在LB层的好处是网关不需要处理TLS开销，性能更好。但需要注意LB到网关之间是明文传输，如果不在同一机房需要额外加密。放在网关层的好处是端到端加密，但Go的TLS握手会消耗CPU。

我们选择在LB层终止TLS，LB到网关之间走VPC内网。证书管理使用cert-manager自动签发和轮换。

Token鉴权流程

WebSocket的鉴权与普通HTTP API不同，因为握手完成后就不再有HTTP请求了。Token只能在握手阶段传递。三种方式：

  1. URL参数：wss://gateway.example.com/ws?token=xxx。简单但Token会出现在日志中，有泄露风险。
  2. Header：握手请求中携带Authorization: Bearer xxx。更安全但浏览器原生WebSocket API不支持自定义Header。
  3. 子协议：在Sec-WebSocket-Protocol中传递Token。兼容浏览器但不够标准。

我们的方案是：浏览器客户端用URL参数（但Token使用短期票据，5分钟有效），原生客户端用Header。短期票据通过HTTP API获取，需要带上长期Token验证。

速率限制策略

三层速率限制防止恶意客户端拖垮网关：

go
package gateway

import (
	"sync"
	"time"
)

// RateLimiter 速率限制器
type RateLimiter struct {
	mu sync.Mutex
	// 单连接限流: 每秒最多N条消息
	connLimit int
	// 单用户限流: 所有设备合计每秒最多N条消息
	userLimit int
	// 全局限流: 整个网关每秒最多N条消息
	globalLimit int

	connCounters   map[string]*SlidingWindow
	userCounters   map[string]*SlidingWindow
	globalCounter  *SlidingWindow
}

// SlidingWindow 滑动窗口计数器
type SlidingWindow struct {
	timestamps []time.Time
	limit      int
	window     time.Duration
}

func (sw *SlidingWindow) Allow() bool {
	now := time.Now()
	cutoff := now.Add(-sw.window)

	// 清理过期时间戳
	idx := 0
	for i, t := range sw.timestamps {
		if t.After(cutoff) {
			idx = i
			break
		}
	}
	sw.timestamps = sw.timestamps[idx:]

	if len(sw.timestamps) >= sw.limit {
		return false
	}

	sw.timestamps = append(sw.timestamps, now)
	return true
}

安全设计不是上线后加的补丁，而是架构设计时就预留的地基。后期补安全，就像在建成的大楼里加装消防通道——能做但代价大。

9.5 DeepSeek辅助架构设计 ​

这一节我分享一个真实的经历：在设计WebSocket网关架构的过程中，我用DeepSeek做了辅助分析。不是让AI替代我做设计，而是用它来做方案的交叉验证和盲区发现。

9.5.1 为什么要用AI辅助 ​

架构设计最怕的不是方案不够好，而是"你不知道你不知道"。一个人设计架构的时候，思维盲区是必然存在的。让团队成员review可以发现问题，但团队成员也有思维惯性。

DeepSeek这类大语言模型的优势在于：它见过大量的架构方案，能快速指出你没想到的问题。它的劣势在于：它不了解你的具体业务场景，给出的建议可能不切实际。

所以正确的用法是：AI做发散，人做收敛。AI提可能性，人做决策。

AI不是架构师的替代品，而是架构师的"第二双眼睛"。它看到的和你看到的叠加在一起，才是更完整的图景。

9.5.2 架构Review的Prompt设计 ​

我用了结构化的prompt来让DeepSeek做架构review。核心思路是把架构设计的上下文完整地提供给AI，然后让它从特定角度做分析。

你是一个资深的分布式系统架构师，请帮我review以下WebSocket网关架构设计。

## 系统背景
- 业务场景：即时消息(IM)、实时数据推送、协同编辑
- 预期规模：16万在线连接，峰值10万QPS
- 技术栈：Go 1.21, Redis Cluster, Kafka
- 部署环境：Kubernetes, 8节点

## 架构设计
1. 接入层：HAProxy做负载均衡，TLS终止
2. 网关层：8个WebSocket网关节点，每节点支持2万连接
3. 注册中心：Redis Cluster存储连接路由信息
4. 消息总线：Kafka处理跨服务异步消息
5. 存储层：Redis缓存会话信息，MySQL持久化离线消息

## 连接模型
- 每条连接2个goroutine（读/写）
- 连接表用sync.RWMutex保护的map，双索引（connID + userID）
- 心跳间隔30秒，Pong超时10秒
- 注册中心TTL 90秒

## 消息路由
- 点对点：查Redis获取目标连接节点，本地直发或RPC转发
- 群组：查Redis获取成员列表，按节点批量转发
- 广播：本地遍历 + 远程RPC

请从以下角度分析：
1. 单点故障风险
2. 性能瓶颈
3. 数据一致性风险
4. 可扩展性限制
5. 运维盲区

每个问题给出：严重程度(高/中/低)、具体描述、建议方案。

9.5.3 DeepSeek的review结果与我的改进 ​

DeepSeek给出了7个问题，我筛选后采纳了其中5个，以下是具体内容：

问题1（高）：Redis注册中心单点风险

DeepSeek指出：虽然用了Redis Cluster，但连接注册信息全部存在Redis上，如果Redis Cluster整体不可用，网关将无法做跨节点路由。

我的改进：增加本地路由缓存。每条连接注册时同时在本地内存缓存一份路由信息，Redis不可用时降级为本地路由模式（只能发本地连接的消息），保证基本可用。

问题2（高）：广播消息的扇出放大

DeepSeek指出：广播消息需要RPC调用所有节点，节点数增加时RPC调用次数线性增长。8个节点时尚可接受，但扩展到50个节点时广播延迟会显著增加。

我的改进：引入广播树结构。节点之间组织成树形拓扑，广播消息沿树传播，将扇出从O(N)降低到O(log N)。短期内8个节点不需要这个优化，但架构上预留了扩展空间。

此外DeepSeek还指出了一个更隐蔽的问题：广播时所有节点同时向各自的所有连接发送消息，会导致瞬间出口带宽激增。如果广播消息较大（比如1KB），10万连接同时收到就是100MB的瞬间出口流量，可能打满网卡带宽。解决方案是对广播消息做速率平滑，每个节点分批发送，每批1000个连接，间隔10毫秒。

问题3（中）：连接表锁竞争

DeepSeek指出：连接表用sync.RWMutex保护，在高并发写入（大量连接同时建立/断开）时可能成为瓶颈。

我的改进：分片锁。将连接表按连接ID哈希分到N个shard（N=CPU核心数），每个shard独立加锁，减少锁竞争。

go
// ShardedConnectionTable 分片连接表
type ShardedConnectionTable struct {
	shards []*connectionShard
	size   int
}

type connectionShard struct {
	mu       sync.RWMutex
	byConnID map[string]*Connection
	byUserID map[string]map[string]*Connection
}

func NewShardedConnectionTable(shardCount, maxConns int) *ShardedConnectionTable {
	shards := make([]*connectionShard, shardCount)
	for i := 0; i < shardCount; i++ {
		shards[i] = &connectionShard{
			byConnID: make(map[string]*Connection),
			byUserID: make(map[string]map[string]*Connection),
		}
	}
	return &ShardedConnectionTable{
		shards: shards,
		size:   shardCount,
	}
}

func (t *ShardedConnectionTable) getShard(connID string) *connectionShard {
	h := fnv.New32a()
	h.Write([]byte(connID))
	return t.shards[h.Sum32()%uint32(t.size)]
}

func (t *ShardedConnectionTable) Add(conn *Connection) error {
	shard := t.getShard(conn.ID)
	shard.mu.Lock()
	defer shard.mu.Unlock()

	shard.byConnID[conn.ID] = conn
	if shard.byUserID[conn.UserID] == nil {
		shard.byUserID[conn.UserID] = make(map[string]*Connection)
	}
	shard.byUserID[conn.UserID][conn.ID] = conn
	return nil
}

func (t *ShardedConnectionTable) Get(connID string) *Connection {
	shard := t.getShard(connID)
	shard.mu.RLock()
	defer shard.mu.RUnlock()
	return shard.byConnID[connID]
}

func (t *ShardedConnectionTable) GetByUserID(userID string) []*Connection {
	// 需要遍历所有shard
	var conns []*Connection
	for _, shard := range t.shards {
		shard.mu.RLock()
		if userConns, ok := shard.byUserID[userID]; ok {
			for _, conn := range userConns {
				conns = append(conns, conn)
			}
		}
		shard.mu.RUnlock()
	}
	return conns
}

问题4（中）：会话恢复期间的消息时序

DeepSeek指出：客户端重连时，会话恢复和消息缓冲取出是两个独立操作，存在竞态条件。如果消息在会话恢复之后、客户端准备好接收之前到达，可能丢失。

我的改进：在会话恢复期间设置一个"恢复中"状态标记，此期间到达的消息同时写入缓冲队列。会话恢复完成后，先投递缓冲队列的消息，再恢复正常消息流。

问题5（低）：Goroutine泄漏风险

DeepSeek指出：每条连接2个goroutine，如果连接异常关闭但goroutine没有被正确回收，2万连接的goroutine泄漏会导致内存持续增长。

我的改进：增加goroutine生命周期管理，确保连接关闭时两个goroutine都能正确退出。使用context.WithCancel替代closeChan，更符合Go的并发模式。

AI的架构review不是"AI说的都对"，而是"AI说的值得听"。5个采纳，2个不采纳，这个比例说明AI的价值在于发散而非决策。

9.5.4 AI辅助的边界 ​

经过这次实践，我总结了AI辅助架构设计的几条原则：

适合AI做的事：

  • 发散性思考：列举可能的问题、方案、风险
  • 模式匹配：从大量已有方案中找到相似场景的最佳实践
  • 文档生成：把架构决策转化为结构化文档
  • 交叉验证：检查设计是否有遗漏

不适合AI做的事：

  • 做技术选型决策：AI不了解你的团队能力、历史债务、业务约束
  • 评估性能：AI给出的性能数字不可信，必须实测
  • 处理业务特殊约束：比如"不能引入Kafka因为运维团队搞不定"
  • 评估成本：AI不了解你的云服务定价和人力成本

AI辅助的正确姿势：

  1. 先自己做设计，形成完整方案
  2. 让AI做review，列出潜在问题
  3. 筛选AI的建议，决定采纳哪些
  4. 让AI帮你把采纳的建议整合到方案中
  5. 自己做最终决策

这五步流程看起来简单，但实际执行时需要注意一个关键问题：AI的review质量高度依赖你提供的上下文质量。如果你只给AI一个模糊的描述，它给出的review也是模糊的。你需要把架构设计文档、容量评估数据、技术约束条件都完整地提供给AI，它才能给出有针对性的建议。

另外，AI的review结果需要你有一定的判断力来筛选。DeepSeek给出的7个问题中，有2个我判断不适用于我们的场景：一个是建议用Raft一致性算法替代Redis注册中心，这在我们的规模下属于过度设计；另一个是建议引入Service Mesh做可观测性，但团队对Istio不熟悉，引入成本太高。这种判断力来自于你对业务场景和团队能力的深入理解，是AI无法替代的。

go
// 架构设计的AI辅助流程（伪代码）
func DesignWithAI(humanDesign *Architecture) *Architecture {
    // 1. 人类设计师出方案
    design := humanDesign
    
    // 2. AI做发散性review
    issues := deepseek.Review(design, ReviewPrompt)
    
    // 3. 人类筛选建议
    for _, issue := range issues {
        if humanAgrees(issue) && fitsContext(issue) {
            design = applyFix(design, issue)
        }
    }
    
    // 4. AI验证修改后的方案
    validation := deepseek.Validate(design)
    
    // 5. 人类做最终决策
    return humanFinalize(design, validation)
}

AI是架构师的工具，不是架构师本身。用好了如虎添翼，用差了画蛇添足。关键在于你知道什么时候该听AI的，什么时候该信自己。

9.6 架构设计文档模板 ​

这一节我把整个架构设计的输出物整理成一个模板，方便你在自己的项目中复用。

9.6.1 架构设计文档结构 ​

以下是一个完整的WebSocket网关架构设计文档大纲：

markdown
# WebSocket网关架构设计文档

## 1. 概述
### 1.1 背景与目标
### 1.2 名词解释
### 1.3 文档范围

## 2. 需求分析
### 2.1 业务场景
### 2.2 功能性需求
### 2.3 非功能性需求
### 2.4 容量评估
### 2.5 约束条件

## 3. 架构设计
### 3.1 整体架构
### 3.2 模块划分
### 3.3 数据流
### 3.4 部署架构
### 3.5 扩展性设计

## 4. 详细设计
### 4.1 连接管理
### 4.2 消息路由
### 4.3 会话管理
### 4.4 心跳机制
### 4.5 安全设计
### 4.6 监控设计

## 5. 关键技术选型
### 5.1 WebSocket库选型
### 5.2 注册中心选型
### 5.3 消息队列选型
### 5.4 序列化协议选型

## 6. 容量规划
### 6.1 单机性能指标
### 6.2 集群规模规划
### 6.3 资源需求估算

## 7. 可靠性设计
### 7.1 故障场景分析
### 7.2 容灾策略
### 7.3 降级方案
### 7.4 限流策略

## 8. 运维设计
### 8.1 部署流程
### 8.2 配置管理
### 8.3 监控告警
### 8.4 日志规范
### 8.5 故障排查

## 9. 演进规划
### 9.1 Phase 1: MVP
### 9.2 Phase 2: 水平扩展
### 9.3 Phase 3: 高可用
### 9.4 Phase 4: 智能化

9.6.2 ADR决策记录模板 ​

架构决策记录（ADR）是记录重要技术决策的工具。在WebSocket网关设计中，我记录了以下几条关键决策：

markdown
# ADR-001: 使用Redis Cluster作为连接注册中心

## 状态
Accepted

## 背景
WebSocket网关需要跨节点路由消息，需要查询用户连接所在节点。
候选方案：etcd、Consul、Redis Cluster、ZooKeeper。

## 决策
选择Redis Cluster。

## 理由
1. 现有基础设施已有Redis Cluster，无需引入新组件
2. Redis读写延迟低（<1ms），满足路由查询需求
3. Redis Cluster提供自动分片和高可用
4. TTL机制天然支持连接超时清理
5. 团队对Redis运维经验丰富

## 代价
1. Redis Cluster不保证强一致性，极端情况下路由信息可能短暂不一致
2. 需要处理Redis不可用时的降级逻辑

# ADR-002: 每条连接使用2个goroutine

## 状态
Accepted

## 背景
WebSocket连接需要同时读取和写入数据。
候选方案：1个goroutine + epoll、2个goroutine（读+写）、3个goroutine（读+写+处理）。

## 决策
选择2个goroutine方案（读+写）。

## 理由
1. 代码简洁，易于理解和维护
2. Go goroutine的开销足够小（初始2KB栈），2万连接约80MB
3. 读写分离避免互斥锁
4. 写goroutine通过channel接收消息，天然支持背压

## 代价
1. 2万连接需要4万goroutine，GC压力需要关注
2. 消息处理在read goroutine中执行，处理慢会阻塞读取

# ADR-003: 消息序列化使用Protocol Buffers

## 状态
Accepted

## 背景
WebSocket消息需要序列化后传输。
候选方案：JSON、Protocol Buffers、MessagePack。

## 决策
选择Protocol Buffers。

## 理由
1. 二进制格式，体积比JSON小30%-50%
2. 序列化/反序列化速度比JSON快5-10倍
3. 强类型，编译时就能发现字段错误
4. 向后兼容性好，新增字段不影响旧版本

## 代价
1. 需要维护.proto文件
2. 调试不如JSON直观
3. 前端需要引入protobuf.js库

好的架构文档不是写给别人看的，而是写给未来的自己看的。半年后你回头看，能快速回忆起"当时为什么这么决策"，这比任何代码注释都有用。

9.7 关键设计决策总结 ​

把这一章的设计决策整理成一张表，方便回顾：

决策项选择核心理由代价
协议WebSocket (RFC 6455)全双工、标准协议、浏览器原生支持需要处理握手、帧解析、心跳
连接模型每连接2 goroutine简洁、高效、读写分离2万连接4万goroutine，需关注GC
连接表分片双索引map减少锁竞争，快速查找按用户ID查找需遍历所有shard
注册中心Redis Cluster低延迟、自动分片、TTL清理不保证强一致性
消息路由本地直发+远程RPC减少中间环节，低延迟节点间RPC需要维护
心跳Ping/Pong 30s间隔标准机制、兼容性好需处理Pong超时
会话恢复Redis缓冲5分钟用户体验好、消息不丢增加Redis存储开销
优雅关闭Close帧+超时强制客户端可感知、不丢消息关闭时间延长10秒
序列化Protocol Buffers体积小、速度快、强类型需维护proto文件

总结 ​

这篇是WebSocket网关模块的开篇，没有写一行业务代码，全在讲协议、需求和架构。但正是这些"虚"的东西，决定了后面代码的"实"。

回顾这一章的核心知识点：

协议层：

  • WebSocket握手是HTTP协议升级，Sec-WebSocket-Key/Accept是协议验证不是安全机制
  • 帧格式包括FIN、opcode、MASK、payload length，理解帧格式是调试的基础
  • 消息分片允许大消息拆分传输，控制帧可以穿插在分片消息之间
  • 心跳机制（Ping/Pong）是长连接保活的必需手段
  • 关闭流程是双向握手，状态码传达关闭原因

业务层：

  • 实时通信分四类场景：IM、数据推送、协同编辑、信令
  • 容量评估需要从连接数、消息QPS、带宽、延迟四个维度量化
  • 功能需求清单覆盖连接管理、消息路由、会话管理、可靠性、运维监控
  • 非功能需求定义了性能、可用性、安全的具体指标

架构层：

  • 三种架构方案的利弊分析，选择"多机+Redis Cluster"的务实方案
  • 网关节点内部模块划分：Accepter、Connection Manager、Router、Auth、Heartbeat
  • 消息路由三种模式：点对点、群组、广播
  • 连接表分片设计减少锁竞争
  • 注册中心用Redis Cluster实现跨节点路由
  • 会话恢复机制支持重连后上下文恢复和消息补投
  • 优雅关闭流程避免消息丢失和重连风暴
  • AI辅助架构review发现5个有效问题

架构设计是"想清楚再动手"的过程。代码可以重构，架构重构的代价是十倍以上。前期多花一周时间在设计上，后期少花一个月填坑，这笔账怎么算都划算。怕浪猫在这一点上吃过太多亏了，每一句建议都是用血泪换来的。

觉得有用？收藏这篇文章，下次设计WebSocket或任何实时通信系统的时候，照着架构设计文档模板和ADR模板填一遍，你会发现思路清晰很多。

你在WebSocket开发中遇到过什么坑？或者对这篇的架构设计有什么不同看法？评论区聊聊，怕浪猫在线接砖。

关注怕浪猫，下期我们进入WebSocket网关的核心功能实现——从握手接入到消息分发，一行一行Go代码手把手实现。

系列进度 9/16 — 下一篇：WebSocket网关核心功能实现

怕浪猫说：从通知平台到权限系统再到WebSocket网关，这个系列的难度在爬坡。WebSocket这块我踩过的坑最多，从协议层面的帧解析到架构层面的跨节点路由，每一个细节都值得深挖。这篇是地基，下篇开始砌墙。打地基的时候别嫌慢，墙砌歪了拆了重砌更痛苦。记住，好的架构设计是演进出来的，不是一次画出来的。保持务实，保持敬畏，我们下篇见。

热爱生活，喜好美食，追求未来！

Copyright © 2026-present 焕然一新组合出版