逻辑实现过程
1. 目标与基础设计
-
目标:实现一个类似 HTTP 请求的阻塞式调用接口(如 await wsRequest(...)),让开发者无需手动处理 WebSocket 的事件回调,而是通过 Promise 和 async/await 获得同步体验。
-
基础设计:
-
创建一个 WebSocketClient 类,封装 WebSocket 的连接和消息处理逻辑。
-
使用 Promise 将 WebSocket 的异步消息机制转化为请求-响应模型。
-
为每个请求生成唯一的 requestId,用于匹配客户端请求和服务器响应。
-
2. WebSocket 连接管理
-
初始化连接:
-
在构造函数中实例化 WebSocket 对象,并绑定基本事件监听(onopen、onmessage、onerror、onclose)。
-
维护一个 isConnected 状态,跟踪连接是否可用。
-
-
事件处理:
-
onopen:标记连接成功,触发队列处理。
-
onmessage:解析收到的消息,提取 requestId 和 data,通过 requestMap 找到对应的 Promise 并完成它。
-
onerror 和 onclose:标记连接失败,清理未完成请求并尝试重连。
-
3. wsRequest 方法的核心逻辑
-
请求发送:
-
生成唯一 requestId,与请求数据组成一个对象(如 { requestId, data })。
-
检查连接状态(isConnected 和 ws.readyState),确保请求只在连接可用时发送。
-
将请求数据序列化为 JSON 并通过 ws.send 发送。
-
-
Promise 封装:
-
创建一个 Promise,将其 resolve 函数存储到 requestMap 中,以 requestId 为键。
-
当收到服务器响应时,通过 requestId 找到对应的 resolve 并执行。
-
-
超时处理:
-
为每个请求设置超时计时器(默认 5 秒),超时后移除 requestMap 中的记录并触发重试或失败。
-
4. 消息队列与重连机制
-
消息队列:
-
当连接不可用时,将请求(包括数据和 Promise 的 resolve/reject)存入 messageQueue。
-
连接恢复后,逐步处理队列中的请求。
-
-
重连机制:
-
如果连接断开,通过 reconnect 方法延迟(默认 1 秒)尝试重新连接。
-
重连成功后触发 flushQueue,处理积压的请求。
-
5. 队列处理的异步优化
-
初始实现:
-
使用 while 循环同步处理队列,可能导致主线程阻塞。
-
-
优化为异步:
-
使用 setTimeout 递归调用 flushQueue,每次处理一个请求后将控制权交还给事件循环。
-
在请求完成后(.finally),异步触发下一次处理。
-
6. 错误与清理
-
错误处理:
-
连接错误或关闭时,通过 rejectAllPending 拒绝所有未完成请求。
-
超时或重试超限时,返回明确的错误信息。
-
-
资源清理:
-
在超时或请求完成时清理 requestMap,避免内存泄漏。
-
提供 close 方法,手动关闭 WebSocket 连接。
-
7. 最终接口
-
用户通过 await wsClient.wsRequest({data: xxx}) 调用,获得类似 HTTP 请求的阻塞式体验。
-
支持配置选项(如超时时间、重试次数、重试间隔),增强灵活性。
考虑的问题与解决方案
1. WebSocket 的异步特性
-
问题:WebSocket 是事件驱动的,没有内置的请求-响应匹配机制。
-
解决方案:
-
使用 requestId 标识每个请求,服务器返回时带上相同的 requestId。
-
用 Promise 封装请求,监听 onmessage 时根据 requestId 完成对应的 Promise。
-
2. 连接状态的不确定性
-
问题:WebSocket 可能未连接、断开或出错,导致请求失败。
-
解决方案:
-
检查 readyState 和 isConnected,未连接时将请求加入队列并尝试重连。
-
提供重试机制(默认 3 次),确保网络不稳定时仍有机会成功。
-
3. 请求超时的风险
-
问题:服务器可能延迟响应,导致 Promise 无限等待。
-
解决方案:
-
为每个请求设置超时(默认 5 秒),超时后触发重试或失败。
-
在 requestMap 中动态更新 resolve,确保超时后仍能正确清理。
-
4. 主线程阻塞
-
问题:使用 while 处理队列可能阻塞主线程,影响页面响应性。
-
解决方案:
-
改为 setTimeout 异步递归处理队列,每次只处理一个请求,避免同步循环。
-
5. 服务器配合
-
问题:WebSocket 本身不保证响应格式,依赖服务器正确实现。
-
解决方案:
-
假设服务器返回 JSON 格式的消息,包含 requestId 和 data。
-
文档中明确说明服务器端需配合返回匹配的 requestId。
-
6. 资源管理与内存泄漏
-
问题:未完成的请求可能堆积,导致内存泄漏。
-
解决方案:
-
在超时、连接关闭或错误时清理 requestMap。
-
队列处理失败时移除对应项,避免无限积压。
-
7. 用户体验与灵活性
-
问题:开发者需要简单易用的接口,同时支持自定义配置。
-
解决方案:
-
提供阻塞式接口(await wsRequest),隐藏复杂的事件逻辑。
-
通过 options 支持配置超时、重试次数等参数。
-
class WebSocketClient {constructor(url, options = {}) {this.url = url;this.ws = null;this.requestMap = new Map(); // 存储 requestId 和 Promise resolverthis.messageQueue = []; // 未连接时的消息队列this.isConnected = false;this.options = {maxRetries: options.maxRetries || 3, // 默认最大重试次数retryDelay: options.retryDelay || 1000, // 默认重试间隔 1 秒timeout: options.timeout || 5000, // 默认超时 5 秒};this.connect(); // 初始化连接}// 建立 WebSocket 连接connect() {this.ws = new WebSocket(this.url);this.ws.onopen = () => {console.log('WebSocket 连接已建立');this.isConnected = true;this.flushQueue(); // 连接成功后发送队列中的消息};this.ws.onmessage = (event) => {const response = JSON.parse(event.data);const { requestId, data } = response;const resolver = this.requestMap.get(requestId);if (resolver) {resolver(data);this.requestMap.delete(requestId);}};this.ws.onerror = (error) => {console.error('WebSocket 错误:', error);this.isConnected = false;};this.ws.onclose = () => {console.log('WebSocket 连接已关闭');this.isConnected = false;this.rejectAllPending('WebSocket 连接关闭');};}// 发送请求并返回 PromisewsRequest(data, retryCount = 0) {return new Promise((resolve, reject) => {// 生成唯一 requestIdconst requestId = Date.now() + Math.random().toString(36).slice(2);const request = { requestId, data };// 检查连接状态if (!this.isConnected || this.ws.readyState !== WebSocket.OPEN) {if (retryCount < this.options.maxRetries) {console.log(`WebSocket 未连接,第 ${retryCount + 1} 次重试...`);this.messageQueue.push({ request, resolve, reject, retryCount: retryCount + 1 });this.reconnect();return;}return reject(new Error('WebSocket 未连接且重试次数已达上限'));}// 存储 resolverthis.requestMap.set(requestId, resolve);// 发送请求this.ws.send(JSON.stringify(request));// 设置超时const timeoutId = setTimeout(() => {if (this.requestMap.has(requestId)) {if (retryCount < this.options.maxRetries) {console.log(`请求超时,第 ${retryCount + 1} 次重试...`);this.requestMap.delete(requestId);this.wsRequest(data, retryCount + 1).then(resolve).catch(reject);} else {reject(new Error('请求超时且重试次数已达上限'));this.requestMap.delete(requestId);}}}, this.options.timeout);// 清理超时时绑定this.requestMap.set(requestId, (value) => {clearTimeout(timeoutId);resolve(value);});});}// 重连逻辑reconnect() {if (!this.isConnected) {setTimeout(() => {console.log('尝试重连...');this.connect();}, this.options.retryDelay);}}// 使用 setTimeout 异步处理消息队列flushQueue() {if (!this.isConnected || this.messageQueue.length === 0) return;const { request, resolve, reject, retryCount } = this.messageQueue.shift();this.wsRequest(request.data, retryCount).then(resolve).catch(reject).finally(() => {// 递归调用,确保队列逐步处理setTimeout(() => this.flushQueue(), 0);});}// 拒绝所有未完成的请求rejectAllPending(errorMessage) {this.requestMap.forEach((resolver, requestId) => {resolver(Promise.reject(new Error(errorMessage)));this.requestMap.delete(requestId);});}// 关闭连接close() {if (this.ws) {this.ws.close();}}
}// 使用示例
async function testWebSocket() {const wsClient = new WebSocketClient('ws://example.com/socket', {maxRetries: 3,retryDelay: 1000,timeout: 5000,});try {// 发送阻塞式请求const response = await wsClient.wsRequest({ action: 'getData', value: 'xxx' });console.log('收到响应:', response);// 模拟未连接时发送请求wsClient.ws.close(); // 手动关闭连接以测试重试和队列const response2 = await wsClient.wsRequest({ action: 'update', value: 'yyy' });console.log('收到响应:', response2);} catch (error) {console.error('请求失败:', error);} finally {wsClient.close();}
}testWebSocket();