WebSocket 大规模推送架构：连接管理、网关与广播

# /etc/sysctl.conf —— 支撑海量长连接
# 单进程可打开的文件描述符上限
fs.file-max = 2097152

# 本地端口范围（主动外连场景）
net.ipv4.ip_local_port_range = 1024 65535

# 全连接队列
net.core.somaxconn = 65535

# TCP 超时与 keepalive
net.ipv4.tcp_keepalive_time = 7200
net.ipv4.tcp_fin_timeout = 15

# 允许更多 TIME_WAIT 复用
net.ipv4.tcp_tw_reuse = 1

# 单进程软/硬限制
# /etc/security/limits.conf
# * soft nofile 1048576
# * hard nofile 1048576

// Go 中的连接会话（内存友好设计）
type Session struct {
    Conn    *websocket.Conn
    UserID  string
    GroupID string   // 所属频道
    LastBeat int64   // 最后心跳（unix 秒），用于超时清理
    WriteCh chan []byte  // 每连接写缓冲
    // 避免为每个连接重复分配，使用 sync.Pool 复用读写缓冲
}

// O(1) 路由：userID → Session
var sessions = map[string]*Session{}
var sessionsMu sync.RWMutex

// 连接建立后注册到 Redis 目录（跨节点路由用）
func registerSession(ctx context.Context, userID, nodeID, connID string) error {
    key := "ws:route:" + userID
    // 5 分钟 TTL + 心跳续期，节点故障后自动过期
    return redisClient.SetEX(ctx, key, nodeID+"|"+connID, 5*time.Minute).Err()
}

# Nginx 作为 WebSocket 卸载层：只终结 TLS 与升级
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

upstream ws_cluster {
    server 10.0.0.11:9000;
    server 10.0.0.12:9000;
    server 10.0.0.13:9000;
    # 长连接节点：一致性哈希保证同一用户固定同一节点
    hash $arg_user_id consistent;
}

server {
    listen 443 ssl http2;
    server_name push.example.com;

    ssl_certificate     /etc/nginx/tls/fullchain.pem;
    ssl_certificate_key /etc/nginx/tls/privkey.pem;

    location /ws/ {
        proxy_pass http://ws_cluster;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header X-User-ID $arg_user_id;

        proxy_read_timeout 3600s;   # 长连接不因空闲被切断
        proxy_send_timeout 3600s;
        proxy_connect_timeout 5s;
    }
}

# Redis 维护 组 → 用户 的订阅关系（set）
# 上线：SADD ws:group:room-1001 user_123
# 下线：SREM ws:group:room-1001 user_123
# 广播：SMEMBERS ws:group:room-1001 → 逐个投递

# 用户 → 组 的反向索引（用于清理）
# SADD ws:user:user_123 room-1001

// 网关节点消费总线消息并扇出到本地连接
func (gw *Gateway) consumeAndFanout(ctx context.Context) {
    for msg := range gw.kafkaReader.Messages() {
        var push PushMsg
        protojson.Unmarshal(msg.Value, &push)

        switch push.Scope {
        case ScopeBroadcast:        // 全量广播
            gw.sessions.Range(func(_ string, s *Session) bool {
                s.Enqueue(push.Payload)     // 异步写
                return true
            })
        case ScopeGroup:            // 组播
            for _, s := range gw.groupIndex[push.Group] {
                s.Enqueue(push.Payload)
            }
        case ScopeUser:             // 单播
            if s := gw.sessions[push.UserID]; s != nil {
                s.Enqueue(push.Payload)
            }
        }
    }
}

// 服务端心跳：每 30s Ping，60s 无响应则清理
func (gw *Gateway) startHeartbeat() {
    go func() {
        for range time.Tick(30 * time.Second) {
            now := time.Now().Unix()
            gw.sessions.Range(func(id string, s *Session) bool {
                if now-s.LastBeat > 60 {
                    s.Conn.Close()          // 判定死链，触发清理
                    gw.removeSession(s)
                    return true
                }
                if err := s.Conn.WriteControl(websocket.PingMessage,
                    nil, 5*time.Second); err != nil {
                    s.Conn.Close()
                }
                return true
            })
        }
    }()
}

// 客户端指数退避重连（jitter）
function connect() {
  const ws = new WebSocket('wss://push.example.com/ws');
  let retries = 0;

  ws.onclose = () => {
    const delay = Math.min(1000 * 2 ** retries, 30000) +
                  Math.random() * 1000;       // 加抖动防惊群
    retries++;
    setTimeout(connect, delay);
  };
  ws.onopen = () => { retries = 0; };
}

// 服务端为每个用户维护单调递增 seq，消息带 seq 下发
type PushMsg struct {
    Seq     uint64 `json:"seq"`
    Payload []byte `json:"payload"`
}

// 客户端把已收到的最大 seq 回传给服务端
// 断线重连时：GET /ws/offline?after=1042
// 服务端返回 (1042, 1050] 之间的离线消息

// 每连接有界写通道 + 非阻塞入队
const maxPending = 1024

func (s *Session) Enqueue(msg []byte) bool {
    select {
    case s.WriteCh <- msg:
        return true
    default:            // 缓冲已满
        // 策略：丢弃并标记该连接需重连补偿 / 主动断开
        s.conn.Close()
        return false
    }
}

// 写协程统一串行 flush，避免并发写 socket
func (s *Session) writeLoop() {
    for msg := range s.WriteCh {
        s.conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
        if err := s.conn.WriteMessage(websocket.BinaryMessage, msg); err != nil {
            return
        }
    }
}

// SSE 一行接入
const es = new EventSource('/api/events?stream=ai');
es.onmessage = (e) => renderChunk(e.data);
es.onerror = () => es.close();  // 默认自动重连