WebSocket 客户端开发：浏览器实战

在前两篇文章中,我们深入探讨了 WebSocket 的基础原理和服务端开发。今天,让我们把目光转向客户端,看看如何在浏览器中构建强大的 WebSocket 客户端。我曾在一个实时协作项目中,通过优化 WebSocket 客户端的重连机制和消息队列,使得用户即使在网络不稳定的情况下也能保持良好的体验。

基础架构设计

一个可靠的 WebSocket 客户端需要考虑以下几个关键点：

连接管理
消息处理
重连机制
心跳检测
错误处理

让我们从基础架构开始：

// websocket-client.js
class WebSocketClient {
  constructor(url, options = {}) {
    this.url = url
    this.options = {
      reconnectInterval: 1000,
      maxReconnectAttempts: 5,
      heartbeatInterval: 30000,
      ...options
    }

    this.handlers = new Map()
    this.messageQueue = []
    this.reconnectAttempts = 0
    this.isConnecting = false
    this.heartbeatTimer = null

    this.connect()
  }

  // 建立连接
  connect() {
    if (this.isConnecting) return

    this.isConnecting = true

    try {
      this.ws = new WebSocket(this.url)
      this.setupEventListeners()
    } catch (error) {
      console.error('Connection error:', error)
      this.handleReconnect()
    }
  }

  // 设置事件监听
  setupEventListeners() {
    this.ws.onopen = () => {
      console.log('Connected to WebSocket server')
      this.isConnecting = false
      this.reconnectAttempts = 0

      // 启动心跳
      this.startHeartbeat()

      // 处理队列中的消息
      this.processMessageQueue()

      // 触发连接事件
      this.emit('connect')
    }

    this.ws.onclose = () => {
      console.log('Disconnected from WebSocket server')
      this.cleanup()
      this.handleReconnect()

      // 触发断开事件
      this.emit('disconnect')
    }

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error)
      this.emit('error', error)
    }

    this.ws.onmessage = (event) => {
      try {
        const message = JSON.parse(event.data)
        this.handleMessage(message)
      } catch (error) {
        console.error('Message parsing error:', error)
      }
    }
  }

  // 发送消息
  send(type, data) {
    const message = {
      type,
      data,
      id: this.generateMessageId(),
      timestamp: Date.now()
    }

    if (this.isConnected()) {
      this.sendMessage(message)
    } else {
      // 添加到消息队列
      this.messageQueue.push(message)
    }

    return message.id
  }

  // 实际发送消息
  sendMessage(message) {
    try {
      this.ws.send(JSON.stringify(message))
      this.emit('sent', message)
    } catch (error) {
      console.error('Send error:', error)
      // 添加到重试队列
      this.messageQueue.push(message)
    }
  }

  // 处理消息队列
  processMessageQueue() {
    while (this.messageQueue.length > 0) {
      const message = this.messageQueue.shift()
      this.sendMessage(message)
    }
  }

  // 处理收到的消息
  handleMessage(message) {
    // 处理心跳响应
    if (message.type === 'pong') {
      this.handleHeartbeatResponse()
      return
    }

    // 触发消息事件
    this.emit('message', message)

    // 调用特定类型的处理器
    const handler = this.handlers.get(message.type)
    if (handler) {
      handler(message.data)
    }
  }

  // 注册消息处理器
  on(type, handler) {
    this.handlers.set(type, handler)
  }

  // 移除消息处理器
  off(type) {
    this.handlers.delete(type)
  }

  // 触发事件
  emit(event, data) {
    const handler = this.handlers.get(event)
    if (handler) {
      handler(data)
    }
  }

  // 处理重连
  handleReconnect() {
    if (this.reconnectAttempts >= this.options.maxReconnectAttempts) {
      console.log('Max reconnection attempts reached')
      this.emit('reconnect_failed')
      return
    }

    this.reconnectAttempts++
    const delay = this.calculateReconnectDelay()

    console.log(`Reconnecting in ${delay}ms... (attempt ${this.reconnectAttempts})`)

    setTimeout(() => {
      this.connect()
    }, delay)

    this.emit('reconnecting', {
      attempt: this.reconnectAttempts,
      delay
    })
  }

  // 计算重连延迟
  calculateReconnectDelay() {
    // 使用指数退避算法
    return Math.min(
      1000 * Math.pow(2, this.reconnectAttempts),
      30000
    )
  }

  // 启动心跳
  startHeartbeat() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer)
    }

    this.heartbeatTimer = setInterval(() => {
      this.sendHeartbeat()
    }, this.options.heartbeatInterval)
  }

  // 发送心跳
  sendHeartbeat() {
    if (this.isConnected()) {
      this.send('ping', {
        timestamp: Date.now()
      })
    }
  }

  // 处理心跳响应
  handleHeartbeatResponse() {
    // 可以在这里添加心跳延迟统计等逻辑
  }

  // 清理资源
  cleanup() {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer)
      this.heartbeatTimer = null
    }
  }

  // 检查连接状态
  isConnected() {
    return this.ws && this.ws.readyState === WebSocket.OPEN
  }

  // 生成消息ID
  generateMessageId() {
    return Math.random().toString(36).substr(2, 9)
  }

  // 关闭连接
  close() {
    if (this.ws) {
      this.ws.close()
    }
    this.cleanup()
  }
}

状态管理

实现可靠的状态管理机制：

// state-manager.js
class StateManager {
  constructor() {
    this.state = {
      connected: false,
      reconnecting: false,
      messageCount: 0,
      lastMessageTime: null,
      errors: [],
      latency: 0
    }

    this.listeners = new Set()
  }

  // 更新状态
  update(changes) {
    const oldState = { ...this.state }
    this.state = {
      ...this.state,
      ...changes
    }

    this.notifyListeners(oldState)
  }

  // 获取状态
  get() {
    return { ...this.state }
  }

  // 添加监听器
  addListener(listener) {
    this.listeners.add(listener)
  }

  // 移除监听器
  removeListener(listener) {
    this.listeners.delete(listener)
  }

  // 通知监听器
  notifyListeners(oldState) {
    this.listeners.forEach(listener => {
      listener(this.state, oldState)
    })
  }

  // 重置状态
  reset() {
    this.update({
      connected: false,
      reconnecting: false,
      messageCount: 0,
      lastMessageTime: null,
      errors: [],
      latency: 0
    })
  }
}

消息队列管理

实现可靠的消息队列管理：

// message-queue.js
class MessageQueue {
  constructor(options = {}) {
    this.options = {
      maxSize: 1000,
      retryLimit: 3,
      retryDelay: 1000,
      ...options
    }

    this.queue = []
    this.processing = false
  }

  // 添加消息
  add(message) {
    if (this.queue.length >= this.options.maxSize) {
      this.handleQueueOverflow()
    }

    this.queue.push({
      message,
      attempts: 0,
      timestamp: Date.now()
    })

    this.process()
  }

  // 处理队列
  async process() {
    if (this.processing) return

    this.processing = true

    while (this.queue.length > 0) {
      const item = this.queue[0]

      try {
        await this.sendMessage(item.message)
        this.queue.shift()
      } catch (error) {
        if (item.attempts >= this.options.retryLimit) {
          this.queue.shift()
          this.handleFailedMessage(item)
        } else {
          item.attempts++
          await this.wait(this.options.retryDelay)
        }
      }
    }

    this.processing = false
  }

  // 发送消息
  async sendMessage(message) {
    // 实现具体的发送逻辑
    return new Promise((resolve, reject) => {
      // 模拟发送
      if (Math.random() > 0.5) {
        resolve()
      } else {
        reject(new Error('Send failed'))
      }
    })
  }

  // 处理队列溢出
  handleQueueOverflow() {
    // 可以选择丢弃最旧的消息
    this.queue.shift()
  }

  // 处理失败的消息
  handleFailedMessage(item) {
    console.error('Message failed after max retries:', item)
  }

  // 等待指定时间
  wait(ms) {
    return new Promise(resolve => setTimeout(resolve, ms))
  }

  // 清空队列
  clear() {
    this.queue = []
    this.processing = false
  }

  // 获取队列状态
  getStatus() {
    return {
      size: this.queue.length,
      processing: this.processing
    }
  }
}

使用示例

让我们看看如何使用这个 WebSocket 客户端：

// 创建客户端实例
const client = new WebSocketClient('ws://localhost:8080', {
  reconnectInterval: 2000,
  maxReconnectAttempts: 10,
  heartbeatInterval: 20000
})

// 状态管理
const stateManager = new StateManager()

// 消息队列
const messageQueue = new MessageQueue({
  maxSize: 500,
  retryLimit: 5
})

// 注册事件处理器
client.on('connect', () => {
  stateManager.update({ connected: true })
  console.log('Connected!')
})

client.on('disconnect', () => {
  stateManager.update({ connected: false })
  console.log('Disconnected!')
})

client.on('message', (message) => {
  stateManager.update({
    messageCount: stateManager.get().messageCount + 1,
    lastMessageTime: Date.now()
  })

  console.log('Received:', message)
})

client.on('error', (error) => {
  const errors = [...stateManager.get().errors, error]
  stateManager.update({ errors })

  console.error('Error:', error)
})

// 监听状态变化
stateManager.addListener((newState, oldState) => {
  // 更新UI或触发其他操作
  updateUI(newState)
})

// 发送消息
function sendMessage(type, data) {
  // 添加到消息队列
  messageQueue.add({
    type,
    data
  })

  // 通过WebSocket发送
  client.send(type, data)
}

// 更新UI
function updateUI(state) {
  // 实现UI更新逻辑
  console.log('State updated:', state)
}

// 使用示例
sendMessage('chat', {
  text: 'Hello, WebSocket!',
  user: 'Alice'
})