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

心上之秋

于 2025-01-05 15:30:23 发布

阅读量903

点赞数 25

文章标签： websocket 网络协议网络

本文链接：https://blog.csdn.net/qq_35485206/article/details/144954072

版权

在前两篇文章中,我们深入探讨了 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()
  }
}

1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.
73.
74.
75.
76.
77.
78.
79.
80.
81.
82.
83.
84.
85.
86.
87.
88.
89.
90.
91.
92.
93.
94.
95.
96.
97.
98.
99.
100.
101.
102.
103.
104.
105.
106.
107.
108.
109.
110.
111.
112.
113.
114.
115.
116.
117.
118.
119.
120.
121.
122.
123.
124.
125.
126.
127.
128.
129.
130.
131.
132.
133.
134.
135.
136.
137.
138.
139.
140.
141.
142.
143.
144.
145.
146.
147.
148.
149.
150.
151.
152.
153.
154.
155.
156.
157.
158.
159.
160.
161.
162.
163.
164.
165.
166.
167.
168.
169.
170.
171.
172.
173.
174.
175.
176.
177.
178.
179.
180.
181.
182.
183.
184.
185.
186.
187.
188.
189.
190.
191.
192.
193.
194.
195.
196.
197.
198.
199.
200.
201.
202.
203.
204.
205.
206.
207.
208.
209.
210.
211.
212.
213.
214.
215.
216.
217.
218.
219.
220.
221.
222.
223.
224.
225.
226.
227.
228.
229.
230.
231.
232.
233.
234.

状态管理

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

// 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'
})