8.1 WebSocket 基础

8.1 WebSocket 基础

WebSocket协议是现代Web实时通信的基石，它提供了全双工、低延迟的网络通信能力，完美克服了传统HTTP轮询的效率问题。本节将深入探讨WebSocket的核心概念、API使用和最佳实践。

协议概述与核心优势

1. 与传统HTTP对比

2. 协议握手过程

sequenceDiagram
    participant Client
    participant Server
    
    Client->>Server: GET /chat HTTP/1.1
    Client->>Server: Upgrade: websocket
    Client->>Server: Connection: Upgrade
    Client->>Server: Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==
    
    Server-->>Client: HTTP/1.1 101 Switching Protocols
    Server-->>Client: Upgrade: websocket
    Server-->>Client: Connection: Upgrade
    Server-->>Client: Sec-WebSocket-Accept: HSmrc0sMlYUkAGmm5OPpG2HaGWk=

客户端API详解

1. 基础连接管理

// 创建WebSocket连接
const socket = new WebSocket('wss://echo.websocket.org');

// 连接事件处理
socket.onopen = (event) => {
  console.log('连接已建立', event);
  socket.send('Hello Server!');
};

socket.onmessage = (event) => {
  console.log('收到消息:', event.data);
  document.getElementById('messages').innerHTML += `<div>${event.data}</div>`;
};

socket.onclose = (event) => {
  console.log('连接关闭:', event.code, event.reason);
  if (event.wasClean) {
    console.log('连接正常关闭');
  } else {
    console.error('连接异常断开');
    // 尝试重连
    setTimeout(connectWebSocket, 5000);
  }
};

socket.onerror = (error) => {
  console.error('WebSocket错误:', error);
};

2. 高级消息处理

二进制数据传输：

// 发送ArrayBuffer
const buffer = new ArrayBuffer(128);
socket.send(buffer);

// 发送Blob数据
const blob = new Blob(['Binary data'], { type: 'application/octet-stream' });
socket.send(blob);

// 接收二进制数据
socket.onmessage = (event) => {
  if (event.data instanceof ArrayBuffer) {
    const view = new Uint8Array(event.data);
    processBinaryData(view);
  } else {
    console.log('文本消息:', event.data);
  }
};

消息序列化策略：

// 发送JSON数据
function sendMessage(type, payload) {
  if (socket.readyState === WebSocket.OPEN) {
    socket.send(JSON.stringify({
      type,
      data: payload,
      timestamp: Date.now()
    }));
  }
}

// 接收处理
socket.onmessage = (event) => {
  try {
    const message = JSON.parse(event.data);
    switch (message.type) {
      case 'CHAT_MESSAGE':
        displayChatMessage(message.data);
        break;
      case 'SYSTEM_ALERT':
        showSystemAlert(message.data);
        break;
      default:
        console.warn('未知消息类型:', message.type);
    }
  } catch (e) {
    console.error('消息解析失败:', e);
  }
};

服务器端实现要点

1. Node.js实现示例

const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

// 连接管理
wss.on('connection', (ws) => {
  console.log('新客户端连接');
  
  // 消息处理
  ws.on('message', (message) => {
    console.log('收到消息:', message);
    
    // 广播给所有客户端
    wss.clients.forEach(client => {
      if (client.readyState === WebSocket.OPEN) {
        client.send(message);
      }
    });
  });
  
  // 连接关闭
  ws.on('close', () => {
    console.log('客户端断开连接');
  });
  
  // 发送欢迎消息
  ws.send(JSON.stringify({
    type: 'WELCOME',
    data: 'Connected to WebSocket server'
  }));
});

console.log('WebSocket服务器运行在 ws://localhost:8080');

2. 连接状态管理

// 跟踪所有连接
const clients = new Set();

wss.on('connection', (ws) => {
  clients.add(ws);
  
  ws.on('close', () => {
    clients.delete(ws);
    updateOnlineCount();
  });
  
  updateOnlineCount();
  
  function updateOnlineCount() {
    const count = clients.size;
    broadcast({ type: 'USER_COUNT', count });
  }
  
  function broadcast(message) {
    clients.forEach(client => {
      if (client.readyState === WebSocket.OPEN) {
        client.send(JSON.stringify(message));
      }
    });
  }
});

生产环境最佳实践

1. 连接可靠性增强

心跳检测机制：

// 客户端实现
let heartbeatInterval;

function setupHeartbeat() {
  // 每30秒发送心跳
  heartbeatInterval = setInterval(() => {
    if (socket.readyState === WebSocket.OPEN) {
      socket.send(JSON.stringify({ type: 'PING' }));
    }
  }, 30000);
  
  // 服务器应响应PONG
}

socket.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'PONG') {
    // 连接正常
    resetConnectionTimeout();
  }
};

let timeoutId;
function resetConnectionTimeout() {
  clearTimeout(timeoutId);
  // 40秒无响应认为连接已死
  timeoutId = setTimeout(() => {
    socket.close();
  }, 40000);
}

自动重连策略：

let reconnectAttempts = 0;
const maxReconnectAttempts = 5;
const baseDelay = 1000;

function connect() {
  const socket = new WebSocket('wss://example.com/ws');
  
  socket.onclose = () => {
    if (reconnectAttempts < maxReconnectAttempts) {
      const delay = baseDelay * Math.pow(2, reconnectAttempts);
      reconnectAttempts++;
      setTimeout(connect, delay);
    }
  };
  
  socket.onopen = () => {
    reconnectAttempts = 0;
  };
}

2. 安全防护措施

WSS加密连接：

# Nginx配置WebSocket代理
server {
    listen 443 ssl;
    server_name example.com;
    
    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;
    
    location /ws/ {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
    }
}

消息验证与过滤：

// 服务器端消息验证
function isValidMessage(message) {
  // 检查消息类型
  const validTypes = ['CHAT', 'JOIN', 'LEAVE'];
  if (!validTypes.includes(message.type)) {
    return false;
  }
  
  // 检查数据大小
  if (JSON.stringify(message.data).length > 1024) {
    return false;
  }
  
  // 防止XSS
  if (typeof message.data === 'string' && /<script.*?>.*?<\/script>/gi.test(message.data)) {
    return false;
  }
  
  return true;
}

ws.on('message', (data) => {
  try {
    const message = JSON.parse(data);
    if (!isValidMessage(message)) {
      ws.close(1008, 'Invalid message');
      return;
    }
    processMessage(message);
  } catch (e) {
    ws.close(1007, 'Invalid JSON');
  }
});

性能优化技巧

1. 消息压缩策略

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

function sendCompressed(data) {
  const jsonStr = JSON.stringify(data);
  const compressed = pako.deflate(jsonStr);
  socket.send(compressed);
}

// 接收解压
socket.onmessage = (event) => {
  let data;
  if (event.data instanceof Blob) {
    data = pako.inflate(new Uint8Array(await event.data.arrayBuffer()));
  } else if (event.data instanceof ArrayBuffer) {
    data = pako.inflate(new Uint8Array(event.data));
  } else {
    try {
      data = JSON.parse(event.data);
    } catch (e) {
      console.error('解析失败:', e);
    }
  }
  processMessage(data);
};

2. 批量消息处理

// 客户端批量发送
let messageQueue = [];
let batchTimer;

function enqueueMessage(message) {
  messageQueue.push(message);
  
  if (!batchTimer) {
    batchTimer = setTimeout(sendBatch, 100); // 100ms批处理窗口
  }
}

function sendBatch() {
  if (socket.readyState === WebSocket.OPEN && messageQueue.length) {
    socket.send(JSON.stringify(messageQueue));
    messageQueue = [];
  }
  batchTimer = null;
}

// 服务器批量处理
ws.on('message', (data) => {
  const messages = JSON.parse(data);
  if (Array.isArray(messages)) {
    messages.forEach(processSingleMessage);
  } else {
    processSingleMessage(messages);
  }
});

实际应用案例

1. 实时协作编辑器

// 客户端实现
class CollaborativeEditor {
  constructor(socket) {
    this.socket = socket;
    this.setupSocket();
    this.setupEditor();
  }
  
  setupSocket() {
    this.socket.onmessage = (event) => {
      const { type, data } = JSON.parse(event.data);
      
      switch (type) {
        case 'DOCUMENT_UPDATE':
          this.applyRemoteUpdate(data);
          break;
        case 'USER_LIST':
          this.updateUserList(data);
          break;
      }
    };
  }
  
  setupEditor() {
    this.editor = document.getElementById('editor');
    this.editor.addEventListener('input', this.debounceLocalUpdate.bind(this));
  }
  
  debounceLocalUpdate() {
    clearTimeout(this.debounceTimer);
    this.debounceTimer = setTimeout(() => {
      this.sendLocalUpdate();
    }, 300);
  }
  
  sendLocalUpdate() {
    const content = this.editor.value;
    this.socket.send(JSON.stringify({
      type: 'DOCUMENT_UPDATE',
      data: { content }
    }));
  }
  
  applyRemoteUpdate(update) {
    // 解决冲突的算法可以根据需求实现
    this.editor.value = update.content;
  }
  
  updateUserList(users) {
    document.getElementById('userCount').textContent = users.length;
  }
}

2. 实时游戏状态同步

// 游戏客户端
class GameClient {
  constructor() {
    this.socket = new WebSocket('wss://game.example.com');
    this.gameState = {};
    this.pendingInputs = [];
    
    this.setupNetwork();
    this.startGameLoop();
  }
  
  setupNetwork() {
    this.socket.onmessage = (event) => {
      const snapshot = JSON.parse(event.data);
      this.applyGameState(snapshot);
    };
  }
  
  applyGameState(snapshot) {
    // 服务器权威状态
    this.gameState = snapshot;
  }
  
  startGameLoop() {
    setInterval(() => {
      // 客户端预测
      this.processInputs();
      
      // 发送输入到服务器
      if (this.pendingInputs.length) {
        this.socket.send(JSON.stringify({
          type: 'PLAYER_INPUT',
          inputs: this.pendingInputs
        }));
        this.pendingInputs = [];
      }
      
      // 渲染当前状态
      this.render();
    }, 1000/60); // 60FPS
  }
  
  processInputs() {
    // 根据输入更新本地预测状态
  }
  
  handlePlayerInput(input) {
    this.pendingInputs.push(input);
    this.processInputs(); // 立即响应
  }
}

WebSocket技术为现代Web应用带来了真正的实时通信能力，在实现时需要注意：

• 始终优先使用安全的WSS协议
• 实现健壮的错误处理和重连机制
• 考虑消息压缩和批处理优化性能
• 设计可扩展的消息协议格式
• 注意安全防护和输入验证

对于更复杂的场景，可以考虑以下扩展方向：

• 结合WebRTC实现P2P通信
• 使用Socket.IO等库简化开发
• 集成Protobuf等高效序列化格式
• 实现自定义的QoS保证机制