Vue 3 与 WebSockets 集群应用

概述

WebSockets 提供了浏览器和服务器之间的全双工通信通道，是构建实时应用的关键技术。在大规模应用场景中，单个 WebSocket 服务器无法处理大量并发连接，因此需要构建 WebSockets 集群。本集将深入探讨 Vue 3 与 WebSockets 集群的集成，包括核心概念、集群架构、负载均衡、会话管理、消息广播、故障转移等高级特性，帮助开发者构建高可用、可扩展的实时应用。

核心知识点

1. WebSockets 集群基础

1.1 设计理念

• 高可用性 - 确保 WebSocket 服务在部分节点故障时仍能正常运行
• 可扩展性 - 支持水平扩展，根据负载动态添加或移除节点
• 负载均衡 - 将客户端连接均匀分布到不同的服务器节点
• 会话一致性 - 确保客户端在不同节点间切换时保持会话状态
• 消息可靠性 - 保证消息在集群中可靠传递，不丢失、不重复

1.2 核心概念

• WebSocket 服务器节点 - 单个运行 WebSocket 服务的服务器实例
• 负载均衡器 - 分发客户端连接到不同服务器节点的设备或软件
• 会话存储 - 存储客户端会话信息的共享存储系统
• 消息总线 - 在集群节点间传递消息的中间件
• 心跳机制 - 检测连接状态，维持活跃连接
• 故障转移 - 在节点故障时将流量转移到其他健康节点

1.3 集群架构

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   Vue 应用       │     │  负载均衡器      │     │ WebSocket 节点 1 │
└────────┬────────┘     └────────┬────────┘     └────────┬────────┘
         │                       │                       │
         │ 1. 建立 WebSocket 连接 │ 2. 分发连接请求      │
         ├───────────────────────►│───────────────────────►│
         │                       │                       │
         │                       │ 3. 分发连接请求      │
         │                       ├───────────────────────►│
         │                       │                       │ WebSocket 节点 2
         │                       │ 3. 分发连接请求      │
         │                       ├───────────────────────►│
         │                       │                       │
         │                       │     ┌─────────────┐     │ WebSocket 节点 3
         │ 4. 实时数据通信        │     │  消息总线    │     │
         │◄───────────────────────┤◄────┤             │◄────┤
         │                       │     │             │     │
         │                       │     └─────────────┘     │
         │                       │                         │
         │                       │     ┌─────────────┐     │
         │                       │     │ 会话存储     │     │
         │                       │     └─────────────┘     │
         │                       │                         │
┌────────▼────────┐     ┌────────▼────────┐     ┌────────▼────────┐
│   更新 UI        │     │ 分发客户端请求   │     │ 处理 WebSocket  │
└─────────────────┘     └─────────────────┘     └─────────────────┘

2. WebSockets 集群实现

2.1 客户端实现

// src/composables/useWebSocketCluster.js
import { ref, onMounted, onUnmounted } from 'vue'

export function useWebSocketCluster(url, options = {}) {
  const socket = ref(null)
  const isConnected = ref(false)
  const messages = ref([])
  const connectionError = ref(null)
  const reconnectAttempts = ref(0)
  const maxReconnectAttempts = options.maxReconnectAttempts || 5
  const reconnectDelay = options.reconnectDelay || 1000
  const heartbeatInterval = options.heartbeatInterval || 30000
  let heartbeatTimer = null
  let reconnectTimer = null

  // 建立连接
  const connect = () => {
    try {
      socket.value = new WebSocket(url)
      
      socket.value.onopen = () => {
        console.log('WebSocket 连接已建立')
        isConnected.value = true
        reconnectAttempts.value = 0
        startHeartbeat()
        if (options.onOpen) {
          options.onOpen()
        }
      }
      
      socket.value.onmessage = (event) => {
        const message = JSON.parse(event.data)
        messages.value.push(message)
        if (options.onMessage) {
          options.onMessage(message)
        }
      }
      
      socket.value.onclose = (event) => {
        console.log('WebSocket 连接已关闭:', event.code, event.reason)
        isConnected.value = false
        stopHeartbeat()
        if (options.onClose) {
          options.onClose(event)
        }
        if (!event.wasClean) {
          attemptReconnect()
        }
      }
      
      socket.value.onerror = (error) => {
        console.error('WebSocket 错误:', error)
        connectionError.value = error
        if (options.onError) {
          options.onError(error)
        }
      }
    } catch (error) {
      console.error('WebSocket 连接失败:', error)
      connectionError.value = error
      attemptReconnect()
    }
  }

  // 发送消息
  const send = (data) => {
    if (socket.value && isConnected.value) {
      socket.value.send(JSON.stringify(data))
      return true
    }
    return false
  }

  // 关闭连接
  const close = (code = 1000, reason = 'Normal closure') => {
    if (socket.value) {
      socket.value.close(code, reason)
    }
    stopReconnect()
  }

  // 心跳机制
  const startHeartbeat = () => {
    stopHeartbeat()
    heartbeatTimer = setInterval(() => {
      send({ type: 'heartbeat', timestamp: Date.now() })
    }, heartbeatInterval)
  }

  const stopHeartbeat = () => {
    if (heartbeatTimer) {
      clearInterval(heartbeatTimer)
      heartbeatTimer = null
    }
  }

  // 重连机制
  const attemptReconnect = () => {
    if (reconnectAttempts.value < maxReconnectAttempts) {
      reconnectAttempts.value++
      const delay = reconnectDelay * Math.pow(2, reconnectAttempts.value - 1) // 指数退避
      console.log(`尝试第 ${reconnectAttempts.value} 次重连，延迟 ${delay}ms`)
      
      reconnectTimer = setTimeout(() => {
        connect()
      }, delay)
    } else {
      console.error('达到最大重连次数，停止重连')
      if (options.onMaxReconnectAttempts) {
        options.onMaxReconnectAttempts()
      }
    }
  }

  const stopReconnect = () => {
    if (reconnectTimer) {
      clearTimeout(reconnectTimer)
      reconnectTimer = null
    }
  }

  // 清理资源
  const cleanup = () => {
    close()
    stopHeartbeat()
    stopReconnect()
  }

  // 生命周期钩子
  onMounted(() => {
    connect()
  })

  onUnmounted(() => {
    cleanup()
  })

  return {
    socket,
    isConnected,
    messages,
    connectionError,
    reconnectAttempts,
    connect,
    send,
    close
  }
}

2.2 客户端使用示例

<template>
  <div class="websocket-cluster-demo">
    <h2>WebSocket 集群实时通信</h2>
    <div class="connection-status">
      <span :class="{ connected: isConnected, disconnected: !isConnected }">
        {{ isConnected ? '已连接' : '未连接' }}
      </span>
      <button @click="connect" v-if="!isConnected">重新连接</button>
      <button @click="close" v-else>关闭连接</button>
    </div>
    
    <div class="messages">
      <h3>消息列表</h3>
      <ul>
        <li v-for="(msg, index) in messages" :key="index">
          <strong>{{ msg.type }}:</strong> {{ JSON.stringify(msg.data) }}
        </li>
      </ul>
    </div>
    
    <div class="message-input">
      <input v-model="message" placeholder="输入消息内容" />
      <button @click="sendMessage">发送消息</button>
    </div>
  </div>
</template>

<script setup>
import { ref } from 'vue'
import { useWebSocketCluster } from '@/composables/useWebSocketCluster'

const message = ref('')

// 使用 WebSocket 集群组合式函数
const {
  socket,
  isConnected,
  messages,
  connectionError,
  reconnectAttempts,
  connect,
  send,
  close
} = useWebSocketCluster('wss://your-load-balancer-url/ws', {
  maxReconnectAttempts: 5,
  reconnectDelay: 1000,
  heartbeatInterval: 30000,
  onOpen: () => {
    console.log('WebSocket 集群连接成功')
  },
  onMessage: (msg) => {
    console.log('收到消息:', msg)
  },
  onClose: (event) => {
    console.log('WebSocket 连接关闭:', event)
  },
  onError: (error) => {
    console.error('WebSocket 错误:', error)
  },
  onMaxReconnectAttempts: () => {
    console.error('达到最大重连次数')
  }
})

// 发送消息
const sendMessage = () => {
  if (message.value.trim()) {
    send({
      type: 'chat',
      data: {
        content: message.value,
        timestamp: Date.now(),
        userId: 'user-123'
      }
    })
    message.value = ''
  }
}
</script>

<style scoped>
.websocket-cluster-demo {
  max-width: 800px;
  margin: 0 auto;
  padding: 20px;
}

.connection-status {
  display: flex;
  align-items: center;
  gap: 10px;
  margin-bottom: 20px;
}

.connected {
  color: green;
  font-weight: bold;
}

.disconnected {
  color: red;
  font-weight: bold;
}

.messages {
  margin: 20px 0;
  padding: 15px;
  border: 1px solid #ddd;
  border-radius: 4px;
  max-height: 300px;
  overflow-y: auto;
}

.messages ul {
  list-style: none;
  padding: 0;
}

.messages li {
  padding: 8px;
  border-bottom: 1px solid #eee;
}

.message-input {
  display: flex;
  gap: 10px;
}

.message-input input {
  flex: 1;
  padding: 8px;
  border: 1px solid #ddd;
  border-radius: 4px;
}

.message-input button {
  padding: 8px 16px;
  background-color: #42b983;
  color: white;
  border: none;
  border-radius: 4px;
  cursor: pointer;
}

.message-input button:hover {
  background-color: #359968;
}
</style>

### 3. 服务器端实现

#### 3.1 Node.js 服务器基础架构

```javascript
// server.js
const http = require('http')
const WebSocket = require('ws')
const Redis = require('ioredis')
const express = require('express')

const app = express()
const server = http.createServer(app)
const wss = new WebSocket.Server({ server })

// 连接到 Redis 消息总线
const redis = new Redis({
  host: process.env.REDIS_HOST || 'localhost',
  port: process.env.REDIS_PORT || 6379
})

// 存储当前连接的客户端
const clients = new Map()

// WebSocket 连接处理
wss.on('connection', (ws, req) => {
  // 生成唯一客户端 ID
  const clientId = `client-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`
  
  // 存储客户端连接
  clients.set(clientId, ws)
  
  console.log(`客户端 ${clientId} 已连接，当前连接数: ${clients.size}`)
  
  // 订阅 Redis 频道，接收来自其他节点的消息
  const subscribeChannel = 'websocket-cluster-messages'
  redis.subscribe(subscribeChannel)
  
  // 处理 Redis 消息
  redis.on('message', (channel, message) => {
    if (channel === subscribeChannel) {
      try {
        const parsedMessage = JSON.parse(message)
        // 发送消息给当前客户端
        ws.send(JSON.stringify(parsedMessage))
      } catch (error) {
        console.error('解析 Redis 消息失败:', error)
      }
    }
  })
  
  // 处理客户端消息
  ws.on('message', (message) => {
    try {
      const parsedMessage = JSON.parse(message)
      console.log(`收到客户端 ${clientId} 的消息:`, parsedMessage)
      
      // 处理不同类型的消息
      switch (parsedMessage.type) {
        case 'heartbeat':
          // 回复心跳
          ws.send(JSON.stringify({ 
            type: 'heartbeat-ack', 
            timestamp: Date.now() 
          }))
          break
          
        case 'chat':
          // 广播聊天消息到所有节点
          redis.publish(subscribeChannel, JSON.stringify(parsedMessage))
          break
          
        case 'private-message':
          // 处理私聊消息
          handlePrivateMessage(parsedMessage, clientId)
          break
          
        default:
          console.log('未知消息类型:', parsedMessage.type)
      }
    } catch (error) {
      console.error('解析客户端消息失败:', error)
    }
  })
  
  // 处理连接关闭
  ws.on('close', () => {
    // 从客户端映射中移除
    clients.delete(clientId)
    // 取消 Redis 订阅
    redis.unsubscribe(subscribeChannel)
    console.log(`客户端 ${clientId} 已断开连接，当前连接数: ${clients.size}`)
  })
  
  // 处理连接错误
  ws.on('error', (error) => {
    console.error(`客户端 ${clientId} 错误:`, error)
  })
})

// 处理私聊消息
function handlePrivateMessage(message, senderId) {
  const { recipientId, data } = message
  // 检查接收者是否在当前节点
  if (clients.has(recipientId)) {
    const recipientWs = clients.get(recipientId)
    recipientWs.send(JSON.stringify({
      type: 'private-message',
      data: {
        ...data,
        senderId
      }
    }))
  } else {
    // 如果接收者不在当前节点，通过 Redis 转发
    redis.publish('websocket-cluster-private-messages', JSON.stringify({
      ...message,
      senderId
    }))
  }
}

// 启动服务器
const PORT = process.env.PORT || 3000
server.listen(PORT, () => {
  console.log(`WebSocket 服务器运行在端口 ${PORT}`)
})

3.2 会话管理

// 使用 Redis 存储会话信息
const redis = require('ioredis')

// 会话存储类
class SessionStore {
  constructor() {
    this.redis = new Redis({
      host: process.env.REDIS_HOST || 'localhost',
      port: process.env.REDIS_PORT || 6379
    })
  }
  
  // 保存会话信息
  async saveSession(clientId, sessionData) {
    const key = `session:${clientId}`
    await this.redis.set(key, JSON.stringify(sessionData))
    // 设置会话过期时间（24小时）
    await this.redis.expire(key, 86400)
  }
  
  // 获取会话信息
  async getSession(clientId) {
    const key = `session:${clientId}`
    const sessionData = await this.redis.get(key)
    return sessionData ? JSON.parse(sessionData) : null
  }
  
  // 更新会话信息
  async updateSession(clientId, updates) {
    const session = await this.getSession(clientId)
    if (session) {
      const updatedSession = { ...session, ...updates }
      await this.saveSession(clientId, updatedSession)
      return updatedSession
    }
    return null
  }
  
  // 删除会话信息
  async deleteSession(clientId) {
    const key = `session:${clientId}`
    await this.redis.del(key)
  }
  
  // 获取所有会话
  async getAllSessions() {
    const keys = await this.redis.keys('session:*')
    if (keys.length === 0) return []
    
    const sessions = await this.redis.mget(keys)
    return sessions.map(session => JSON.parse(session))
  }
}

module.exports = new SessionStore()

4. 负载均衡配置

4.1 Nginx 负载均衡配置

# nginx.conf
upstream websocket_cluster {
    # WebSocket 服务器节点
    server ws-node-1:3000 weight=1 max_fails=3 fail_timeout=30s;
    server ws-node-2:3000 weight=1 max_fails=3 fail_timeout=30s;
    server ws-node-3:3000 weight=1 max_fails=3 fail_timeout=30s;
    
    # 使用 IP 哈希策略，确保同一客户端始终连接到同一节点
    ip_hash;
}

server {
    listen 80;
    server_name your-domain.com;
    
    # 重定向 HTTP 到 HTTPS
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    server_name your-domain.com;
    
    # SSL 配置
    ssl_certificate /path/to/ssl/certificate.pem;
    ssl_certificate_key /path/to/ssl/private.key;
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_ciphers HIGH:!aNULL:!MD5;
    
    location / {
        root /path/to/your/vue/app;
        index index.html;
        try_files $uri $uri/ /index.html;
    }
    
    # WebSocket 代理配置
    location /ws {
        proxy_pass http://websocket_cluster;
        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;
        
        # 禁用缓冲
        proxy_buffering off;
    }
}

4.2 Docker 容器化部署

# Dockerfile
FROM node:16-alpine

WORKDIR /app

# 复制依赖文件
COPY package*.json ./

# 安装依赖
RUN npm ci --only=production

# 复制应用代码
COPY . .

# 暴露端口
EXPOSE 3000

# 启动应用
CMD ["node", "server.js"]

# docker-compose.yml
version: '3.8'

services:
  # Redis 消息总线
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    restart: always
  
  # WebSocket 服务器节点 1
  ws-node-1:
    build: .
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - PORT=3000
    restart: always
  
  # WebSocket 服务器节点 2
  ws-node-2:
    build: .
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - PORT=3000
    restart: always
  
  # WebSocket 服务器节点 3
  ws-node-3:
    build: .
    depends_on:
      - redis
    environment:
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - PORT=3000
    restart: always
  
  # Nginx 负载均衡器
  nginx:
    image: nginx:alpine
    depends_on:
      - ws-node-1
      - ws-node-2
      - ws-node-3
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./ssl:/etc/nginx/ssl:ro
    restart: always

volumes:
  redis-data:
    driver: local

5. 消息广播与路由

5.1 发布/订阅模式

// 广播消息到所有节点
function broadcast(message) {
  // 通过 Redis 发布消息
  redis.publish('websocket-cluster-messages', JSON.stringify(message))
}

// 订阅特定频道
function subscribeToChannel(channel, callback) {
  redis.subscribe(channel)
  redis.on('message', (ch, msg) => {
    if (ch === channel) {
      try {
        const parsedMessage = JSON.parse(msg)
        callback(parsedMessage)
      } catch (error) {
        console.error('解析订阅消息失败:', error)
      }
    }
  })
}

// 向特定用户发送消息
async function sendToUser(userId, message) {
  // 先检查用户是否在当前节点
  if (clients.has(userId)) {
    const ws = clients.get(userId)
    ws.send(JSON.stringify(message))
    return true
  }
  
  // 如果不在当前节点，通过 Redis 转发
  await redis.publish('websocket-cluster-direct-messages', JSON.stringify({
    userId,
    message
  }))
  return false
}

5.2 消息路由策略

1. 广播路由 - 将消息发送给所有连接的客户端
2. 单播路由 - 将消息发送给特定客户端
3. 组播路由 - 将消息发送给特定组的客户端
4. 扇形路由 - 将消息发送给特定节点的所有客户端

6. 故障转移与高可用性

6.1 健康检查机制

// 定期检查 Redis 连接
setInterval(() => {
  redis.ping((err, result) => {
    if (err) {
      console.error('Redis 连接错误:', err)
      // 尝试重新连接
      redis.connect()
    }
  })
}, 10000)

// 定期检查其他 WebSocket 节点健康状态
async function checkNodeHealth() {
  const nodes = ['ws-node-1', 'ws-node-2', 'ws-node-3']
  
  for (const node of nodes) {
    try {
      // 发送 HTTP 请求检查节点健康状态
      const response = await fetch(`http://${node}:3000/health`)
      if (response.ok) {
        console.log(`节点 ${node} 健康`)
      } else {
        console.warn(`节点 ${node} 不健康: ${response.status}`)
      }
    } catch (error) {
      console.error(`检查节点 ${node} 健康状态失败:`, error)
    }
  }
}

// 每 30 秒检查一次节点健康状态
setInterval(checkNodeHealth, 30000)

6.2 优雅关闭与重启

// 处理优雅关闭
process.on('SIGTERM', gracefulShutdown)
process.on('SIGINT', gracefulShutdown)

function gracefulShutdown() {
  console.log('接收到关闭信号，开始优雅关闭...')
  
  // 停止接收新连接
  wss.close(() => {
    console.log('WebSocket 服务器已关闭')
    
    // 关闭 Redis 连接
    redis.quit(() => {
      console.log('Redis 连接已关闭')
      process.exit(0)
    })
  })
  
  // 设置超时，强制关闭
  setTimeout(() => {
    console.error('优雅关闭超时，强制退出')
    process.exit(1)
  }, 10000)
}

7. 性能优化

7.1 客户端优化

1. 减少重连频率 - 使用指数退避算法
2. 压缩消息大小 - 使用 JSON 压缩或二进制协议
3. 批量发送消息 - 合并多个小消息
4. 合理设置心跳间隔 - 避免过于频繁的心跳
5. 使用 WebSocket 子协议 - 优化消息格式

7.2 服务器端优化

1. 使用连接池 - 优化数据库和 Redis 连接
2. 限制消息大小 - 防止过大消息占用资源
3. 使用流式处理 - 处理大量并发连接
4. 优化内存使用 - 及时清理无效连接
5. 使用集群模式 - 充分利用多核 CPU

// 使用 Node.js 集群模块
const cluster = require('cluster')
const os = require('os')

if (cluster.isMaster) {
  // 主进程
  const numCPUs = os.cpus().length
  
  console.log(`主进程 ${process.pid} 正在运行`)
  
  // 衍生工作进程
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork()
  }
  
  // 处理工作进程退出
  cluster.on('exit', (worker, code, signal) => {
    console.log(`工作进程 ${worker.process.pid} 已退出`)
    // 重启工作进程
    cluster.fork()
  })
} else {
  // 工作进程 - 运行 WebSocket 服务器
  // 这里放置之前的 WebSocket 服务器代码
  console.log(`工作进程 ${process.pid} 已启动`)
}

8. 监控与日志

8.1 客户端监控

// 在 useWebSocketCluster 中添加监控事件
const useWebSocketCluster = (url, options = {}) => {
  // ... 现有代码 ...
  
  // 监控指标
  const metrics = ref({
    connectionTime: 0,
    messageCount: 0,
    bytesSent: 0,
    bytesReceived: 0,
    pingLatency: 0
  })
  
  // 记录连接时间
  socket.value.onopen = () => {
    metrics.value.connectionTime = Date.now()
    // ... 现有代码 ...
  }
  
  // 记录消息统计
  socket.value.onmessage = (event) => {
    metrics.value.messageCount++
    metrics.value.bytesReceived += event.data.length
    // ... 现有代码 ...
  }
  
  // 记录发送字节数
  const originalSend = socket.value.send
  socket.value.send = (data) => {
    metrics.value.bytesSent += data.length
    return originalSend.call(socket.value, data)
  }
  
  // ... 现有代码 ...
  
  return {
    // ... 现有返回值 ...
    metrics
  }
}

8.2 服务器端日志

// 使用 Winston 进行日志记录
const winston = require('winston')

const logger = winston.createLogger({
  level: process.env.LOG_LEVEL || 'info',
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.json()
  ),
  transports: [
    new winston.transports.File({ filename: 'error.log', level: 'error' }),
    new winston.transports.File({ filename: 'combined.log' })
  ]
})

// 在开发环境中添加控制台输出
if (process.env.NODE_ENV !== 'production') {
  logger.add(new winston.transports.Console({
    format: winston.format.simple()
  }))
}

// 使用日志记录器
logger.info('WebSocket 服务器启动', { port: PORT })
logger.error('连接错误', { clientId, error })
logger.warn('节点不健康', { node: 'ws-node-1' })

9. 安全性考虑

9.1 WebSocket 安全最佳实践

1. 使用 WSS 协议 - 加密 WebSocket 连接
2. 实现认证机制 - 验证客户端身份
3. 使用 CORS 限制 - 限制允许的来源
4. 实现速率限制 - 防止 DDoS 攻击
5. 验证消息格式 - 防止注入攻击
6. 设置合理的超时时间 - 防止资源耗尽

9.2 认证与授权

// 在 WebSocket 连接时进行认证
wss.on('connection', (ws, req) => {
  // 从请求头获取认证令牌
  const authToken = req.headers['sec-websocket-protocol'] || req.headers.cookie
  
  // 验证令牌
  if (!validateAuthToken(authToken)) {
    ws.close(4001, 'Unauthorized')
    return
  }
  
  // ... 现有连接处理代码 ...
})

// 验证认证令牌
function validateAuthToken(token) {
  try {
    // 实现令牌验证逻辑
    // 例如，使用 JWT 验证
    const decoded = jwt.verify(token, process.env.JWT_SECRET)
    return decoded
  } catch (error) {
    return false
  }
}

10. 部署与运维

10.1 CI/CD 流水线

# .github/workflows/deploy.yml
name: Deploy WebSocket Cluster

on:
  push:
    branches: [ main ]

jobs:
  build-and-deploy:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Node.js
      uses: actions/setup-node@v3
      with:
        node-version: '16'
    
    - name: Install dependencies
      run: npm ci
    
    - name: Build Docker images
      run: |
        docker build -t your-docker-registry/ws-server:latest .
    
    - name: Login to Docker Registry
      uses: docker/login-action@v2
      with:
        username: ${{ secrets.DOCKER_USERNAME }}
        password: ${{ secrets.DOCKER_PASSWORD }}
    
    - name: Push Docker images
      run: |
        docker push your-docker-registry/ws-server:latest
    
    - name: Deploy to Kubernetes
      uses: appleboy/kubectl-action@v0.1.4
      with:
        kubeconfig: ${{ secrets.KUBECONFIG }}
        command: apply -f k8s/

10.2 Kubernetes 部署

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ws-server
  labels:
    app: ws-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ws-server
  template:
    metadata:
      labels:
        app: ws-server
    spec:
      containers:
      - name: ws-server
        image: your-docker-registry/ws-server:latest
        ports:
        - containerPort: 3000
        env:
        - name: REDIS_HOST
          value: redis-master
        - name: REDIS_PORT
          value: "6379"
        - name: NODE_ENV
          value: production
        livenessProbe:
          httpGet:
            path: /health
            port: 3000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 3000
          initialDelaySeconds: 5
          periodSeconds: 5

---
# k8s/service.yaml
apiVersion: v1
kind: Service
metadata:
  name: ws-server
  labels:
    app: ws-server
spec:
  selector:
    app: ws-server
  ports:
  - port: 80
    targetPort: 3000
  type: ClusterIP

---
# k8s/ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: ws-ingress
  annotations:
    nginx.ingress.kubernetes.io/rewrite-target: /
    nginx.ingress.kubernetes.io/proxy-read-timeout: "86400"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "86400"
    nginx.ingress.kubernetes.io/websocket-services: "ws-server"
spec:
  tls:
  - hosts:
    - your-domain.com
    secretName: ws-tls
  rules:
  - host: your-domain.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: ws-server
            port:
              number: 80

总结

Vue 3 与 WebSockets 集群的集成是构建高可用、可扩展实时应用的关键技术。通过本文的学习，我们掌握了：

1. WebSockets 集群的核心概念和架构设计
2. Vue 3 客户端 WebSocket 集群连接管理
3. Node.js 服务器端 WebSocket 集群实现
4. Redis 消息总线用于节点间通信
5. 负载均衡配置与会话管理
6. 故障转移与高可用性设计
7. 性能优化与监控日志
8. 安全性考虑与最佳实践
9. 部署与运维策略

通过合理设计 WebSockets 集群架构，结合 Vue 3 的响应式系统和组合式 API，可以构建出能够处理大量并发连接、具有高可用性和可扩展性的实时应用。在实际项目中，还需要根据具体需求进行调整和优化，选择合适的技术栈和架构模式。

参考资源

1. MDN WebSockets API
2. Vue 3 文档
3. WebSocket 协议规范
4. Redis 发布/订阅文档
5. Nginx WebSocket 代理
6. Node.js 集群模块
7. Kubernetes 文档

实践作业

1. 实现一个基于 Vue 3 和 WebSockets 集群的实时聊天应用
2. 配置 Nginx 负载均衡器，部署至少 3 个 WebSocket 服务器节点
3. 使用 Redis 作为消息总线，实现节点间消息同步
4. 实现客户端重连机制和服务器端健康检查
5. 添加监控和日志功能
6. 部署到 Docker 或 Kubernetes 环境

通过完成这些实践作业，你将能够深入理解 WebSockets 集群的工作原理和实现细节，为构建大规模实时应用打下坚实的基础。