WebSocket 实时通信与长连接架构实战 2026 | 实时应用开发完全指南

WebSocket 实时通信与长连接架构实战

实时通信是现代 Web 应用的核心能力。从聊天室、实时协作到在线游戏和金融行情，WebSocket 都是不可替代的技术。本文将系统讲解 WebSocket 协议原理、服务端实现、客户端封装、集群扩展及生产环境最佳实践。

一、WebSocket 协议原理

1.1 与 HTTP 的对比

特性	HTTP	WebSocket
通信模式	请求-响应	全双工
连接生命周期	短连接（HTTP/1.1 Keep-Alive 有限）	长连接
服务端推送	不支持（SSE 可单向推送）	原生支持
头部开销	每次请求携带完整头	仅首次握手有头
协议	`http://` / `https://`	`ws://` / `wss://`
数据格式	文本为主	文本 + 二进制

1.2 握手过程

客户端 → 服务端（HTTP Upgrade 请求）

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

服务端 → 客户端（101 Switching Protocols）

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

握手完成后，TCP 连接升级为 WebSocket 连接
双方可以随时发送数据帧（Frame）

1.3 数据帧结构

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

Opcode:
  0x0: 延续帧
  0x1: 文本帧
  0x2: 二进制帧
  0x8: 关闭帧
  0x9: Ping 帧
  0xA: Pong 帧

二、Node.js WebSocket 服务端

2.1 使用 ws 库

typescript

import { WebSocketServer, WebSocket } from 'ws'
import http from 'http'

const server = http.createServer()
const wss = new WebSocketServer({ server, path: '/ws' })

// 连接管理
const clients = new Map<string, WebSocket>()

wss.on('connection', (ws, req) => {
  const clientId = generateId()
  clients.set(clientId, ws)

  console.log(`Client connected: ${clientId}, total: ${clients.size}`)

  // 发送欢迎消息
  ws.send(JSON.stringify({
    type: 'welcome',
    clientId,
    timestamp: Date.now()
  }))

  // 接收消息
  ws.on('message', (data) => {
    try {
      const message = JSON.parse(data.toString())
      handleMessage(clientId, message)
    } catch (err) {
      console.error('Invalid message:', err)
    }
  })

  // 关闭连接
  ws.on('close', () => {
    clients.delete(clientId)
    console.log(`Client disconnected: ${clientId}, total: ${clients.size}`)
  })

  // 错误处理
  ws.on('error', (err) => {
    console.error(`WebSocket error [${clientId}]:`, err)
  })
})

function handleMessage(senderId: string, message: any) {
  switch (message.type) {
    case 'chat':
      // 广播聊天消息
      broadcast({
        type: 'chat',
        from: senderId,
        content: message.content,
        timestamp: Date.now()
      })
      break

    case 'private':
      // 私聊
      const target = clients.get(message.to)
      if (target && target.readyState === WebSocket.OPEN) {
        target.send(JSON.stringify({
          type: 'private',
          from: senderId,
          content: message.content,
          timestamp: Date.now()
        }))
      }
      break
  }
}

function broadcast(message: object) {
  const data = JSON.stringify(message)
  clients.forEach((ws) => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(data)
    }
  })
}

server.listen(3000, () => {
  console.log('WebSocket server running on ws://localhost:3000/ws')
})

2.2 心跳保活机制

typescript

// 心跳检测：定期发送 Ping，清理无响应的连接
class HeartbeatManager {
  private pingInterval: NodeJS.Timeout
  private clients = new Map<WebSocket, boolean>()

  constructor(private wss: WebSocketServer, private interval = 30000) {
    this.start()
  }

  private start() {
    this.pingInterval = setInterval(() => {
      this.wss.clients.forEach((ws) => {
        if (ws.readyState === WebSocket.OPEN) {
          // 检查上次 Pong 是否收到
          if (this.clients.get(ws) === false) {
            ws.terminate()
            return
          }

          // 标记为未响应，发送 Ping
          this.clients.set(ws, false)
          ws.ping()
        }
      })
    }, this.interval)

    // 监听 Pong 响应
    this.wss.on('connection', (ws) => {
      this.clients.set(ws, true)

      ws.on('pong', () => {
        this.clients.set(ws, true)
      })

      ws.on('close', () => {
        this.clients.delete(ws)
      })
    })
  }

  stop() {
    clearInterval(this.pingInterval)
  }
}

// 使用
const heartbeat = new HeartbeatManager(wss, 30000)

三、Socket.IO 框架

3.1 服务端

typescript

import { Server } from 'socket.io'
import { createServer } from 'http'

const httpServer = createServer()
const io = new Server(httpServer, {
  cors: {
    origin: ['https://example.com', 'http://localhost:5173'],
    credentials: true
  },
  // 适配器配置（集群模式）
  adapter: undefined, // 后面用 Redis Adapter
})

// 命名空间（Namespace）：隔离不同业务
const chatNs = io.of('/chat')
const adminNs = io.of('/admin')

// 中间件：认证
chatNs.use((socket, next) => {
  const token = socket.handshake.auth.token
  if (!token) {
    return next(new Error('Authentication required'))
  }

  try {
    const user = verifyToken(token)
    socket.data.user = user
    next()
  } catch {
    next(new Error('Invalid token'))
  }
})

// 房间（Room）：群组通信
chatNs.on('connection', (socket) => {
  const user = socket.data.user

  console.log(`User connected: ${user.name}`)

  // 加入房间
  socket.on('join-room', (roomId: string) => {
    socket.join(roomId)
    socket.to(roomId).emit('user-joined', {
      userId: user.id,
      name: user.name,
      timestamp: Date.now()
    })
  })

  // 离开房间
  socket.on('leave-room', (roomId: string) => {
    socket.leave(roomId)
    socket.to(roomId).emit('user-left', {
      userId: user.id,
      timestamp: Date.now()
    })
  })

  // 发送消息到房间
  socket.on('send-message', (data: { roomId: string; content: string }) => {
    chatNs.to(data.roomId).emit('new-message', {
      from: user.id,
      name: user.name,
      content: data.content,
      timestamp: Date.now()
    })
  })

  // 输入状态
  socket.on('typing', (roomId: string) => {
    socket.to(roomId).emit('user-typing', { userId: user.id })
  })

  socket.on('stop-typing', (roomId: string) => {
    socket.to(roomId).emit('user-stop-typing', { userId: user.id })
  })

  // 断开连接
  socket.on('disconnect', (reason) => {
    console.log(`User disconnected: ${user.name}, reason: ${reason}`)
  })
})

httpServer.listen(3000, () => {
  console.log('Socket.IO server running on port 3000')
})

3.2 客户端

typescript

import { io, Socket } from 'socket.io-client'

class ChatClient {
  private socket: Socket
  private reconnectAttempts = 0

  constructor(url: string, token: string) {
    this.socket = io(url, {
      auth: { token },
      transports: ['websocket'],  // 仅使用 WebSocket
      reconnection: true,
      reconnectionAttempts: 10,
      reconnectionDelay: 1000,
      reconnectionDelayMax: 5000,
    })

    this.setupListeners()
  }

  private setupListeners() {
    this.socket.on('connect', () => {
      console.log('Connected:', this.socket.id)
      this.reconnectAttempts = 0
    })

    this.socket.on('disconnect', (reason) => {
      console.log('Disconnected:', reason)
    })

    this.socket.on('connect_error', (err) => {
      console.error('Connection error:', err.message)
      this.reconnectAttempts++
    })

    // 重连成功
    this.socket.io.on('reconnect', (attempt) => {
      console.log(`Reconnected after ${attempt} attempts`)
    })
  }

  joinRoom(roomId: string) {
    this.socket.emit('join-room', roomId)
  }

  sendMessage(roomId: string, content: string) {
    this.socket.emit('send-message', { roomId, content })
  }

  onMessage(callback: (data: any) => void) {
    this.socket.on('new-message', callback)
  }

  onUserJoined(callback: (data: any) => void) {
    this.socket.on('user-joined', callback)
  }

  onUserLeft(callback: (data: any) => void) {
    this.socket.on('user-left', callback)
  }

  disconnect() {
    this.socket.disconnect()
  }
}

// 使用
const client = new ChatClient('https://api.example.com/chat', jwtToken)
client.joinRoom('room-123')
client.sendMessage('room-123', 'Hello World!')
client.onMessage((data) => {
  console.log(`${data.name}: ${data.content}`)
})

四、Vue 3 集成

4.1 Composable 封装

typescript

// composables/useWebSocket.ts
import { ref, onMounted, onUnmounted } from 'vue'

export function useWebSocket(url: string) {
  const isConnected = ref(false)
  const messages = ref<any[]>([])
  const error = ref<string | null>(null)

  let ws: WebSocket | null = null
  let reconnectTimer: NodeJS.Timeout | null = null
  let reconnectAttempts = 0
  const MAX_RECONNECT = 5

  function connect() {
    ws = new WebSocket(url)

    ws.onopen = () => {
      isConnected.value = true
      error.value = null
      reconnectAttempts = 0
      console.log('WebSocket connected')
    }

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data)
      messages.value.push(data)
    }

    ws.onerror = (err) => {
      error.value = 'WebSocket error'
      console.error('WebSocket error:', err)
    }

    ws.onclose = (event) => {
      isConnected.value = false
      console.log('WebSocket closed:', event.code, event.reason)

      // 自动重连
      if (reconnectAttempts < MAX_RECONNECT) {
        reconnectAttempts++
        const delay = Math.min(1000 * 2 ** reconnectAttempts, 30000)
        console.log(`Reconnecting in ${delay}ms (attempt ${reconnectAttempts})`)

        reconnectTimer = setTimeout(() => connect(), delay)
      }
    }
  }

  function send(data: any) {
    if (ws && ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify(data))
    } else {
      console.warn('WebSocket is not connected')
    }
  }

  function disconnect() {
    if (reconnectTimer) clearTimeout(reconnectTimer)
    if (ws) {
      ws.close()
      ws = null
    }
  }

  onMounted(connect)
  onUnmounted(disconnect)

  return { isConnected, messages, error, send, disconnect }
}

4.2 在组件中使用

vue

<script setup lang="ts">
import { useWebSocket } from '@/composables/useWebSocket'

const { isConnected, messages, send } = useWebSocket('ws://localhost:3000/ws')

const input = ref('')

function handleSend() {
  if (!input.value.trim()) return
  send({ type: 'chat', content: input.value })
  input.value = ''
}
</script>

<template>
  <div>
    <div class="status">
      <span :class="isConnected ? 'online' : 'offline'">
        {{ isConnected ? '在线' : '离线' }}
      </span>
    </div>

    <div class="messages">
      <div v-for="(msg, i) in messages" :key="i" class="message">
        <span class="from">{{ msg.from || 'System' }}:</span>
        <span class="content">{{ msg.content }}</span>
      </div>
    </div>

    <input
      v-model="input"
      @keyup.enter="handleSend"
      placeholder="输入消息..."
    />
    <button @click="handleSend" :disabled="!isConnected">发送</button>
  </div>
</template>

五、集群扩展

5.1 Redis Pub/Sub 适配器

typescript

import { Server } from 'socket.io'
import { createAdapter } from '@socket.io/redis-adapter'
import { createClient } from 'redis'

const io = new Server()

// 创建 Redis 客户端
const pubClient = createClient({ url: 'redis://localhost:6379' })
const subClient = pubClient.duplicate()

await Promise.all([pubClient.connect(), subClient.connect()])

// 使用 Redis 适配器
io.adapter(createAdapter(pubClient, subClient))

// 现在多个 Socket.IO 实例可以共享事件
// Server A 上的客户端发消息 → Server B 上的客户端也能收到

5.2 多实例部署

                    ┌─────────────┐
                    │   Nginx     │
                    │  负载均衡    │
                    └──────┬──────┘
              ┌────────────┼────────────┐
              ↓            ↓            ↓
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ Server A │ │ Server B │ │ Server C │
        │ Socket.IO│ │ Socket.IO│ │ Socket.IO│
        └────┬─────┘ └────┬─────┘ └────┬─────┘
             │            │            │
             └────────────┼────────────┘
                          ↓
                    ┌──────────┐
                    │  Redis   │
                    │ Pub/Sub  │
                    └──────────┘

nginx

# Nginx WebSocket 负载均衡
upstream websocket {
  ip_hash;  # 确保同一客户端连接到同一服务器
  server 10.0.0.1:3000;
  server 10.0.0.2:3000;
  server 10.0.0.3:3000;
}

server {
  listen 443 ssl;
  server_name ws.example.com;

  location /socket.io/ {
    proxy_pass http://websocket;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;

    # WebSocket 超时设置
    proxy_read_timeout 86400s;
    proxy_send_timeout 86400s;
  }
}

5.3 Redis 消息广播

typescript

// 不使用 Socket.IO Adapter，直接用 Redis Pub/Sub
import { createClient } from 'redis'
import { WebSocketServer } from 'ws'

const redis = createClient({ url: 'redis://localhost:6379' })
await redis.connect()

const wss = new WebSocketServer({ port: 3000 })

// 订阅频道
await redis.subscribe('broadcast', (message) => {
  // 收到 Redis 消息后，广播给本实例的所有客户端
  wss.clients.forEach((ws) => {
    if (ws.readyState === ws.OPEN) {
      ws.send(message)
    }
  })
})

// 发布消息到所有实例
function globalBroadcast(message: object) {
  redis.publish('broadcast', JSON.stringify(message))
}

六、实时应用场景

6.1 实时通知系统

typescript

class NotificationService {
  private userSockets = new Map<string, Set<WebSocket>>()

  // 用户上线
  onUserConnect(userId: string, ws: WebSocket) {
    if (!this.userSockets.has(userId)) {
      this.userSockets.set(userId, new Set())
    }
    this.userSockets.get(userId)!.add(ws)
  }

  // 用户下线
  onUserDisconnect(userId: string, ws: WebSocket) {
    this.userSockets.get(userId)?.delete(ws)
    if (this.userSockets.get(userId)?.size === 0) {
      this.userSockets.delete(userId)
    }
  }

  // 发送通知给指定用户
  sendToUser(userId: string, notification: any) {
    const sockets = this.userSockets.get(userId)
    if (!sockets) return false // 用户不在线

    const data = JSON.stringify({
      type: 'notification',
      data: notification
    })

    sockets.forEach(ws => {
      if (ws.readyState === ws.OPEN) {
        ws.send(data)
      }
    })

    return true
  }

  // 检查用户是否在线
  isUserOnline(userId: string): boolean {
    return this.userSockets.has(userId)
  }
}

6.2 实时协作（OT 算法基础）

typescript

class CollaborativeDoc {
  private content = ''
  private version = 0
  private clients = new Map<WebSocket, { userId: string; cursor: number }>()

  // 应用操作
  applyOperation(ws: WebSocket, op: Operation) {
    // 基于 version 检查
    if (op.baseVersion !== this.version) {
      // 需要变换（OT）
      op = this.transform(op, op.baseVersion)
    }

    // 应用操作
    if (op.type === 'insert') {
      this.content = this.content.slice(0, op.position) +
        op.text + this.content.slice(op.position)
    } else if (op.type === 'delete') {
      this.content = this.content.slice(0, op.position) +
        this.content.slice(op.position + op.length)
    }

    this.version++

    // 广播给其他客户端
    const data = JSON.stringify({
      type: 'operation',
      operation: op,
      version: this.version
    })

    this.clients.forEach((_, client) => {
      if (client !== ws && client.readyState === WebSocket.OPEN) {
        client.send(data)
      }
    })
  }

  transform(op: Operation, baseVersion: number): Operation {
    // 简化的 OT 变换逻辑
    // 实际中需要更复杂的变换算法
    return op
  }
}

七、安全与性能

7.1 认证与授权

typescript

// 方式 1：URL 参数（不推荐，会出现在日志中）
// ws://example.com/ws?token=xxx

// 方式 2：Sec-WebSocket-Protocol 头
const ws = new WebSocket('wss://example.com/ws', ['bearer', jwtToken])

// 方式 3：Cookie（推荐）
// 浏览器会自动携带同源 Cookie
// 前端无需额外配置

// 服务端验证
wss.on('connection', (ws, req) => {
  const cookie = req.headers.cookie
  const token = parseCookie(cookie, 'session')

  if (!token) {
    ws.close(4001, 'Unauthorized')
    return
  }

  try {
    const user = verifyToken(token)
    ws.data = { user }
  } catch {
    ws.close(4001, 'Invalid token')
  }
})

7.2 消息限流

typescript

class RateLimiter {
  private buckets = new Map<string, { count: number; resetTime: number }>()

  check(clientId: string, limit = 50, windowMs = 1000): boolean {
    const now = Date.now()
    const bucket = this.buckets.get(clientId)

    if (!bucket || now > bucket.resetTime) {
      this.buckets.set(clientId, { count: 1, resetTime: now + windowMs })
      return true
    }

    if (bucket.count >= limit) {
      return false
    }

    bucket.count++
    return true
  }
}

const limiter = new RateLimiter()

ws.on('message', (data) => {
  if (!limiter.check(clientId)) {
    ws.close(4003, 'Rate limit exceeded')
    return
  }

  // 处理消息
})

7.3 消息大小限制

typescript

import { WebSocketServer } from 'ws'

const wss = new WebSocketServer({
  port: 3000,
  maxPayload: 1024 * 1024,  // 最大 1MB
})

ws.on('message', (data, isBinary) => {
  if (data.length > 100 * 1024) {
    ws.close(4002, 'Message too large')
    return
  }
  // 处理消息
})

八、监控与运维

8.1 连接统计

typescript

class ConnectionStats {
  private stats = {
    totalConnections: 0,
    currentConnections: 0,
    messagesReceived: 0,
    messagesSent: 0,
    errors: 0,
  }

  onConnect() {
    this.stats.totalConnections++
    this.stats.currentConnections++
  }

  onDisconnect() {
    this.stats.currentConnections--
  }

  onMessageReceived() {
    this.stats.messagesReceived++
  }

  onMessageSent() {
    this.stats.messagesSent++
  }

  getStats() {
    return { ...this.stats, timestamp: Date.now() }
  }
}

// 定期上报到 Prometheus
setInterval(() => {
  const stats = connectionStats.getStats()
  // metrics.gauge('ws_current_connections', stats.currentConnections)
  // metrics.counter('ws_messages_total', stats.messagesReceived)
}, 10000)

8.2 优雅关闭

typescript

async function gracefulShutdown() {
  console.log('Shutting down WebSocket server...')

  // 1. 停止接受新连接
  wss.close()

  // 2. 通知所有客户端
  wss.clients.forEach((ws) => {
    ws.send(JSON.stringify({
      type: 'server-shutdown',
      message: 'Server is shutting down. Please reconnect.',
      retryAfter: 5000
    }))
  })

  // 3. 等待消息发送完成
  await new Promise(resolve => setTimeout(resolve, 2000))

  // 4. 关闭所有连接
  wss.clients.forEach((ws) => {
    ws.close(1001, 'Server shutting down')
  })

  // 5. 关闭 HTTP 服务器
  server.close()
}

process.on('SIGTERM', gracefulShutdown)
process.on('SIGINT', gracefulShutdown)

九、总结

✅ WebSocket 协议原理（握手、数据帧、与 HTTP 对比）
✅ Node.js ws 库实现（连接管理、消息处理、广播）
✅ 心跳保活机制（Ping/Pong、无响应清理）
✅ Socket.IO 框架（命名空间、房间、中间件认证）
✅ 客户端封装（自动重连、指数退避、Vue 3 Composable）
✅ 集群扩展（Redis Pub/Sub Adapter、多实例部署、Nginx 负载均衡）
✅ 实时应用场景（通知系统、实时协作）
✅ 安全防护（认证授权、消息限流、大小限制）
✅ 监控运维（连接统计、优雅关闭）

WebSocket 是构建实时应用的核心技术，掌握从单机到集群的完整方案是全栈工程师的必备能力。

相关阅读：

WebSocket 实时通信与长连接架构实战 2026 | 实时应用开发完全指南 ​

一、WebSocket 协议原理 ​

1.1 与 HTTP 的对比 ​

1.2 握手过程 ​

1.3 数据帧结构 ​

二、Node.js WebSocket 服务端 ​

2.1 使用 ws 库 ​

2.2 心跳保活机制 ​

三、Socket.IO 框架 ​

3.1 服务端 ​

3.2 客户端 ​

四、Vue 3 集成 ​

4.1 Composable 封装 ​

4.2 在组件中使用 ​

五、集群扩展 ​

5.1 Redis Pub/Sub 适配器 ​

5.2 多实例部署 ​

5.3 Redis 消息广播 ​

六、实时应用场景 ​

6.1 实时通知系统 ​

6.2 实时协作（OT 算法基础） ​

七、安全与性能 ​

7.1 认证与授权 ​

7.2 消息限流 ​

7.3 消息大小限制 ​

八、监控与运维 ​

8.1 连接统计 ​

8.2 优雅关闭 ​

九、总结 ​

WebSocket 实时通信与长连接架构实战 2026 | 实时应用开发完全指南

一、WebSocket 协议原理

1.1 与 HTTP 的对比

1.2 握手过程

1.3 数据帧结构

二、Node.js WebSocket 服务端

2.1 使用 ws 库

2.2 心跳保活机制

三、Socket.IO 框架

3.1 服务端

3.2 客户端

四、Vue 3 集成

4.1 Composable 封装

4.2 在组件中使用

五、集群扩展

5.1 Redis Pub/Sub 适配器

5.2 多实例部署

5.3 Redis 消息广播

六、实时应用场景

6.1 实时通知系统

6.2 实时协作（OT 算法基础）

七、安全与性能

7.1 认证与授权

7.2 消息限流

7.3 消息大小限制

八、监控与运维

8.1 连接统计

8.2 优雅关闭

九、总结