WebSocket性能优化策略

WebSocket虽然本身性能优异，但在实际应用中仍需进行多方面的优化以确保最佳性能。

连接管理优化

1. 连接池管理

javascript
class WebSocketPool {
  constructor(url, maxConnections = 10) {
    this.url = url;
    this.maxConnections = maxConnections;
    this.connections = [];
    this.pendingRequests = [];
  }
  
  async getConnection() {
    // 查找空闲连接
    const idleConnection = this.connections.find(conn => conn.busy === false);
    if (idleConnection) {
      idleConnection.busy = true;
      return idleConnection.ws;
    }
    
    // 创建新连接
    if (this.connections.length < this.maxConnections) {
      const ws = await this.createConnection();
      this.connections.push({ ws, busy: true });
      return ws;
    }
    
    // 等待可用连接
    return new Promise(resolve => {
      this.pendingRequests.push(resolve);
    });
  }
  
  releaseConnection(ws) {
    const connection = this.connections.find(conn => conn.ws === ws);
    if (connection) {
      connection.busy = false;
      
      // 处理等待的请求
      if (this.pendingRequests.length > 0) {
        const resolve = this.pendingRequests.shift();
        connection.busy = true;
        resolve(ws);
      }
    }
  }
  
  async createConnection() {
    return new Promise((resolve, reject) => {
      const ws = new WebSocket(this.url);
      ws.onopen = () => resolve(ws);
      ws.onerror = reject;
    });
  }
}

2. 连接复用

javascript
// 单例模式管理全局WebSocket连接
class GlobalWebSocket {
  static instance = null;
  
  constructor(url) {
    if (GlobalWebSocket.instance) {
      return GlobalWebSocket.instance;
    }
    
    this.url = url;
    this.ws = null;
    this.messageHandlers = new Map();
    this.connect();
    
    GlobalWebSocket.instance = this;
  }
  
  connect() {
    this.ws = new WebSocket(this.url);
    
    this.ws.onmessage = (event) => {
      const { type, data } = JSON.parse(event.data);
      const handlers = this.messageHandlers.get(type) || [];
      handlers.forEach(handler => handler(data));
    };
  }
  
  subscribe(type, handler) {
    if (!this.messageHandlers.has(type)) {
      this.messageHandlers.set(type, []);
    }
    this.messageHandlers.get(type).push(handler);
  }
  
  unsubscribe(type, handler) {
    const handlers = this.messageHandlers.get(type) || [];
    const index = handlers.indexOf(handler);
    if (index !== -1) {
      handlers.splice(index, 1);
    }
  }
}

// 使用示例
const ws = new GlobalWebSocket('ws://example.com/socket');
ws.subscribe('chat', (data) => console.log('收到聊天消息:', data));
ws.subscribe('notification', (data) => console.log('收到通知:', data));

消息传输优化

1. 消息压缩

javascript
// 使用pako库进行gzip压缩
import pako from 'pako';

class CompressibleWebSocket {
  constructor(url) {
    this.ws = new WebSocket(url);
    this.setupMessageHandlers();
  }
  
  send(data) {
    const json = JSON.stringify(data);
    const compressed = pako.gzip(json);
    this.ws.send(compressed);
  }
  
  setupMessageHandlers() {
    this.ws.onmessage = (event) => {
      const compressed = new Uint8Array(event.data);
      const decompressed = pako.ungzip(compressed, { to: 'string' });
      const data = JSON.parse(decompressed);
      this.handleMessage(data);
    };
  }
  
  handleMessage(data) {
    // 处理解压后的消息
  }
}

2. 消息批量发送

javascript
class BatchWebSocket {
  constructor(url, batchSize = 10, batchTimeout = 100) {
    this.ws = new WebSocket(url);
    this.batch = [];
    this.batchSize = batchSize;
    this.batchTimeout = batchTimeout;
    this.batchTimer = null;
  }
  
  send(message) {
    this.batch.push(message);
    
    // 达到批量大小立即发送
    if (this.batch.length >= this.batchSize) {
      this.flush();
    } else {
      // 设置超时发送
      this.scheduleFlush();
    }
  }
  
  scheduleFlush() {
    if (this.batchTimer) {
      clearTimeout(this.batchTimer);
    }
    
    this.batchTimer = setTimeout(() => {
      this.flush();
    }, this.batchTimeout);
  }
  
  flush() {
    if (this.batch.length === 0) return;
    
    const batch = [...this.batch];
    this.batch = [];
    
    if (this.batchTimer) {
      clearTimeout(this.batchTimer);
      this.batchTimer = null;
    }
    
    this.ws.send(JSON.stringify({ type: 'batch', messages: batch }));
  }
}

3. 消息优先级

javascript
class PriorityWebSocket {
  constructor(url) {
    this.ws = new WebSocket(url);
    this.highPriorityQueue = [];
    this.normalPriorityQueue = [];
    this.lowPriorityQueue = [];
    this.isSending = false;
    this.setupMessageHandlers();
  }
  
  send(message, priority = 'normal') {
    const queue = this.getQueue(priority);
    queue.push(message);
    this.processQueue();
  }
  
  getQueue(priority) {
    switch (priority) {
      case 'high':
        return this.highPriorityQueue;
      case 'low':
        return this.lowPriorityQueue;
      default:
        return this.normalPriorityQueue;
    }
  }
  
  async processQueue() {
    if (this.isSending) return;
    
    const message = this.getNextMessage();
    if (!message) return;
    
    this.isSending = true;
    
    try {
      await this.sendMessage(message);
    } catch (error) {
      console.error('发送消息失败:', error);
      // 重新加入队列
      this.normalPriorityQueue.unshift(message);
    }
    
    this.isSending = false;
    this.processQueue();
  }
  
  getNextMessage() {
    if (this.highPriorityQueue.length > 0) {
      return this.highPriorityQueue.shift();
    }
    if (this.normalPriorityQueue.length > 0) {
      return this.normalPriorityQueue.shift();
    }
    if (this.lowPriorityQueue.length > 0) {
      return this.lowPriorityQueue.shift();
    }
    return null;
  }
  
  sendMessage(message) {
    return new Promise((resolve, reject) => {
      this.ws.send(JSON.stringify(message));
      resolve();
    });
  }
}

服务器端优化

1. 负载均衡

javascript
// 使用Redis进行WebSocket连接的负载均衡
const Redis = require('ioredis');
const redis = new Redis();

class LoadBalancedWebSocketServer {
  constructor(server) {
    this.wss = new WebSocket.Server({ server });
    this.setupLoadBalancing();
  }
  
  setupLoadBalancing() {
    this.wss.on('connection', (ws, request) => {
      const userId = this.getUserId(request);
      const serverId = this.getServerId(userId);
      
      // 如果连接不在当前服务器，重定向
      if (serverId !== this.currentServerId) {
        ws.close(1000, `Redirect to ${serverId}`);
        return;
      }
      
      // 注册连接
      this.registerConnection(userId, ws);
      
      ws.on('close', () => {
        this.unregisterConnection(userId);
      });
    });
  }
  
  async registerConnection(userId, ws) {
    await redis.hset('websocket_connections', userId, this.currentServerId);
    this.connections.set(userId, ws);
  }
  
  async unregisterConnection(userId) {
    await redis.hdel('websocket_connections', userId);
    this.connections.delete(userId);
  }
  
  async sendToUser(userId, message) {
    const serverId = await redis.hget('websocket_connections', userId);
    
    if (serverId === this.currentServerId) {
      const ws = this.connections.get(userId);
      if (ws && ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(message));
      }
    } else {
      // 转发到目标服务器
      this.forwardToServer(serverId, userId, message);
    }
  }
  
  getServerId(userId) {
    // 一致性哈希算法
    return consistentHash(userId, this.serverList);
  }
}

2. 消息广播优化

javascript
class OptimizedBroadcastServer {
  constructor(server) {
    this.wss = new WebSocket.Server({ server });
    this.rooms = new Map();
    this.setupBroadcast();
  }
  
  setupBroadcast() {
    this.wss.on('connection', (ws, request) => {
      const { roomId } = this.parseRequest(request);
      
      // 加入房间
      this.joinRoom(ws, roomId);
      
      ws.on('message', (data) => {
        const message = JSON.parse(data);
        this.broadcastToRoom(roomId, message, ws);
      });
      
      ws.on('close', () => {
        this.leaveRoom(ws, roomId);
      });
    });
  }
  
  joinRoom(ws, roomId) {
    if (!this.rooms.has(roomId)) {
      this.rooms.set(roomId, new Set());
    }
    this.rooms.get(roomId).add(ws);
  }
  
  leaveRoom(ws, roomId) {
    const room = this.rooms.get(roomId);
    if (room) {
      room.delete(ws);
      if (room.size === 0) {
        this.rooms.delete(roomId);
      }
    }
  }
  
  broadcastToRoom(roomId, message, excludeWs) {
    const room = this.rooms.get(roomId);
    if (!room) return;
    
    const data = JSON.stringify(message);
    
    room.forEach(ws => {
      if (ws !== excludeWs && ws.readyState === WebSocket.OPEN) {
        ws.send(data);
      }
    });
  }
}

监控和调优

1. 性能监控

javascript
class WebSocketMonitor {
  constructor(ws) {
    this.ws = ws;
    this.metrics = {
      messagesSent: 0,
      messagesReceived: 0,
      bytesSent: 0,
      bytesReceived: 0,
      latency: [],
      errors: 0
    };
    this.setupMonitoring();
  }
  
  setupMonitoring() {
    const originalSend = this.ws.send.bind(this.ws);
    
    this.ws.send = (data) => {
      this.metrics.messagesSent++;
      this.metrics.bytesSent += data.length;
      return originalSend(data);
    };
    
    this.ws.onmessage = (event) => {
      this.metrics.messagesReceived++;
      this.metrics.bytesReceived += event.data.length;
      
      // 计算延迟
      if (event.data.timestamp) {
        const latency = Date.now() - event.data.timestamp;
        this.metrics.latency.push(latency);
        
        // 只保留最近100个延迟记录
        if (this.metrics.latency.length > 100) {
          this.metrics.latency.shift();
        }
      }
    };
    
    this.ws.onerror = () => {
      this.metrics.errors++;
    };
  }
  
  getMetrics() {
    return {
      ...this.metrics,
      averageLatency: this.calculateAverageLatency(),
      p99Latency: this.calculateP99Latency()
    };
  }
  
  calculateAverageLatency() {
    if (this.metrics.latency.length === 0) return 0;
    const sum = this.metrics.latency.reduce((a, b) => a + b, 0);
    return sum / this.metrics.latency.length;
  }
  
  calculateP99Latency() {
    if (this.metrics.latency.length === 0) return 0;
    const sorted = [...this.metrics.latency].sort((a, b) => a - b);
    const index = Math.floor(sorted.length * 0.99);
    return sorted[index];
  }
}

最佳实践

1. 连接复用：避免频繁创建和销毁连接
2. 消息压缩：对大消息进行压缩传输
3. 批量发送：合并小消息减少网络往返
4. 优先级队列：重要消息优先发送
5. 负载均衡：分散连接到多个服务器
6. 监控指标：实时监控性能指标
7. 及时调优：根据监控数据调整策略
8. 资源清理：及时清理无用连接和资源