登录
WebSocket服务封装实践：从连接管理到业务功能集成
2025年10月14日 18:33
现代Web应用中的实时通信需求
最近项目中需要将先科的广播系统管理平台移植到系统中，经过不断反复的推翻修改，终于有了这篇文章。主要分享一下在设计websocket过程中的一些小技巧与实践方法。
前言： 在当今的Web应用开发中，实时通信功能已成为许多系统的核心需求。无论是即时聊天、实时数据监控还是广播通知系统，WebSocket技术都扮演着至关重要的角色。然而，直接使用原生的WebSocket API往往会导致代码重复、状态管理混乱和错误处理困难等问题。本文将介绍如何封装一个健壮的WebSocket服务，展示从基础连接管理到高级业务功能集成的最佳实践。
1. 连接管理：建立可靠的双向通信通道
WebSocket服务封装了完整的连接生命周期管理：
// --- 全局变量声明（WebSocket连接状态管理） ---
let connectPromise = null // 核心：保存连接的Promise，用于共享连接状态
let socket = null // 当前的 WebSocket 实例
let connectionStatus = 'disconnected' // 连接状态：'idle'|'connecting'|'connected'|'error'|'disconnected'
let shouldReconnect = true // 控制是否允许自动重连
let reconnectAttempts = 0 // 当前重连尝试次数
// 可配置的重连策略
const MAX_RECONNECT_ATTEMPTS = 5 // 最大重连次数
const RECONNECT_INTERVAL = 3000 // 重连间隔时间（毫秒）
let timer = null // 心跳定时器句柄
连接初始化的关键特性：
- 单例模式实现：避免重复创建连接
- Promise封装：提供异步操作接口
- 自动重连机制：在连接断开时自动尝试恢复
- 状态跟踪：实时监控连接状态
/**
* 初始化 WebSocket 连接（Promise 版本）
*
* 该函数用于建立与 WebSocket 服务器的连接，并在连接成功后自动执行用户登录和心跳机制。
* 使用 Promise 封装连接过程，避免重复创建连接，支持自动重连机制。
*
* @param {string} wsUrl - WebSocket 服务器地址（如：ws://localhost:8080）
* @param {function} onMessage - 可选的消息回调函数（此处未使用，但可扩展）
* @param {number} timeout - 可选：连接超时时间（毫秒），默认 10 秒（当前未实现超时控制）
* @returns {Promise<WebSocket>} - 成功时 resolve，返回 WebSocket 实例（实际 resolve 无参数）
* 失败时 reject，携带错误信息
*/
export const initWebSocket = (wsUrl = 'ws://') => {
// 如果已经有连接或正在连接，则直接返回同一个 Promise
// 避免多次调用 initWebSocket 时创建多个连接
if (connectPromise) {
return connectPromise;
}
// 创建一个新的 Promise 来管理 WebSocket 的连接过程
connectPromise = new Promise((resolve, reject) => {
// 情况 1：如果 WebSocket 已经打开，直接 resolve，无需重复连接
if (socket && socket.readyState === WebSocket.OPEN) {
console.log('WebSocket 已连接，跳过初始化');
return resolve(socket); // 可选择返回 socket 实例
}
// 情况 2：如果 WebSocket 正在连接中，不重复创建，但此处未 reject 或 resolve
// 注意：这里没有处理正在连接的情况，可能导致 Promise 悬挂（潜在问题）
// 建议：可以 reject 或 resolve 等待现有连接完成
if (socket && socket.readyState === WebSocket.CONNECTING) {
console.log('WebSocket 正在连接中...');
// 当前未处理，connectPromise 会一直等待 onopen 或 onclose
// 可优化：监听现有 socket 的 onopen 并 resolve
return; // 不执行后续连接逻辑
}
shouldReconnect = true; // 设置重连标志为 true，表示允许自动重连
reconnectAttempts = 0; // 重置重连尝试次数
socket = new WebSocket(wsUrl); // 创建新的 WebSocket 实例
connectionStatus = 'connecting'; // 更新连接状态
/**
* WebSocket 连接成功打开时触发
*/
socket.onopen = () => {
console.log('WebSocket 连接已建立');
connectionStatus = 'connected';
// 连接成功后尝试用户登录（根据实际业务自行封装）
userLogin('xxx', 'xxx')
.then(() => {
console.log('用户登录成功，开始心跳');
startHeartbeat(); // 登录成功后启动心跳机制，维持连接
resolve(socket); // 登录成功才认为初始化完成，resolve Promise
})
.catch((err) => {
console.error('用户登录失败:', err);
reject(new Error('登录失败')); // 登录失败则 reject
});
};
/**
* 接收到服务器消息时触发
* 假设消息为 JSON 格式
*/
socket.onmessage = (event) => {
let data;
try {
data = JSON.parse(event.data);
// 特殊处理：如果收到心跳响应且 result 不为 0，表示心跳失败，关闭连接
if (data.command === 'heartbeat' && data.result !== 0) {
console.warn('心跳响应失败，关闭连接');
closeWebSocket(); // 调用关闭函数，可能触发重连
}
} catch (e) {
console.error('无法解析消息为 JSON:', event.data);
return; // 解析失败，忽略该消息
}
// 将正常消息通过事件机制广播给其他模块处理
notifyMessage(data);
};
/**
* WebSocket 发生错误时触发
* 注意：error 事件并不一定会导致连接关闭，但应记录日志
*/
socket.onerror = (error) => {
console.error('WebSocket 错误:', error);
connectionStatus = 'error';
// 注意：此处不 reject，因为连接可能仍会通过 onclose 触发重连
};
/**
* WebSocket 连接关闭时触发
* 可能是网络断开、服务端关闭、手动关闭等
*/
socket.onclose = () => {
console.log('WebSocket 连接已关闭');
connectionStatus = 'disconnected';
clearInterval(timer); // 清除心跳定时器
// 判断是否需要自动重连
if (shouldReconnect && reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
reconnectAttempts++;
console.log(`尝试重连... (${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})`);
// 延迟一段时间后尝试重新连接
setTimeout(() => {
connectPromise = null; // 清除旧的 Promise，允许重新调用 initWebSocket
// 递归调用自身进行重连
initWebSocket(wsUrl).catch((err) => {
console.error('重连失败:', err);
});
}, RECONNECT_INTERVAL);
} else {
// 超过最大重连次数或不允许重连
console.warn('达到最大重连次数或已禁止重连，停止重连');
}
};
});
// 返回连接 Promise，调用者可通过 .then().catch() 处理结果
return connectPromise;
};
2. 消息处理：事件总线与命令分发
高效的消息处理是WebSocket服务的核心能力：
import { notifyMessage } from '@/utils/eventBus'; // 引入消息总线
/**
* 接收到服务器消息时触发
* 假设消息为 JSON 格式
*/
socket.onmessage = (event) => {
let data;
try {
data = JSON.parse(event.data);
// 特殊处理：如果收到心跳响应且 result 不为 0，表示心跳失败，关闭连接
if (data.command === 'heartbeat' && data.result !== 0) {
console.warn('心跳响应失败，关闭连接');
closeWebSocket(); // 调用关闭函数，可能触发重连
}
} catch (e) {
console.error('无法解析消息为 JSON:', event.data);
return; // 解析失败，忽略该消息
}
// 将正常消息通过事件机制广播给其他模块处理
notifyMessage(data);
};
利用eventBus事件总线通知全局订阅者，接收消息
// eventBus.js
import mitt from 'mitt'
const subscribers = []
// 订阅消息
export const subscribe = (callback) => {
if (typeof callback === 'function') {
subscribers.push(callback)
}
// 返回取消订阅函数
return () => {
const index = subscribers.indexOf(callback)
if (index > -1) {
subscribers.splice(index, 1)
}
}
}
// 通知所有订阅者
export const notifyMessage = (data) => {
subscribers.forEach((callback) => {
try {
callback(data)
} catch (error) {
console.error('消息回调执行出错:', error)
}
})
}
export default mitt()
消息处理策略：
- JSON数据解析与错误处理
- 特殊解析与错误处理
- 特殊命令的优先处理（如心跳、登录响应）
- 通用消息通过事件总线广播
- 命令路由机制（根据command字段分发处理）
3. Promise封装：管理异步操作
利用Promise管理异步操作使代码更清晰：
// 发送消息的Promise封装
/**
* 发送消息到 WebSocket 服务器（异步安全版本）
*
* 该函数用于向 WebSocket 服务端发送消息。在发送前会确保连接已建立（自动初始化连接），
* 并对消息格式、连接状态进行检查，确保消息可靠发送。
*
* @param {Object|string} message - 要发送的消息内容，通常为对象（将被 JSON.stringify）
* @returns {Promise<boolean>} - 发送成功返回 true，失败返回 false
*
* @example
* const success = await sendMessage({ command: 'chat', data: 'Hello' });
* if (success) {
* console.log('消息发送成功');
* } else {
* console.log('消息发送失败');
* }
*/
export const sendMessage = async (message) => {
// 参数校验：禁止发送空消息
if (!message) {
console.warn('无法发送空消息：message 为 null、undefined 或空值');
return false;
}
try {
// 确保 WebSocket 连接已建立
// 如果尚未连接，initWebSocket 会尝试建立连接并完成登录流程
// 如果连接失败或登录失败，initWebSocket 会 reject，此处捕获并返回 false
await initWebSocket();
} catch (error) {
// initWebSocket 失败（如连接超时、网络问题、登录失败等）
console.warn('WebSocket 初始化失败，无法发送消息:', error.message);
return false;
}
// 再次检查 WebSocket 的当前状态是否为 OPEN（已打开）
// 即使 initWebSocket 成功，网络可能在发送前断开，因此需要二次确认
if (socket && socket.readyState === WebSocket.OPEN) {
try {
// 将消息序列化为 JSON 字符串并发送
socket.send(JSON.stringify(message));
console.log('消息已发送:', message);
return true; // 发送成功
} catch (error) {
// send() 方法在某些异常情况下可能抛出异常（如序列化失败、底层错误）
console.error('WebSocket send() 方法调用失败:', error);
return false;
}
} else {
// WebSocket 未连接或处于 CONNECTING/CLOSING/CLOSED 状态
console.warn('WebSocket 未处于 OPEN 状态，无法发送消息');
return false;
}
};
Promise使用场景：
- 连接初始化：确保连接就绪
- 用户登录：处理认证流程
- 业务操作：如广播寻呼、获取设备信息等
- 错误处理：统一捕获和报告异常
4. 通用方法封装示例
/**
* 获取设备信息
*
* 该函数用于向指定设备发送指令，以获取与指定账户关联的区域（zone）信息。
* 它通过调用 sendMessage 函数发送一个包含设备唯一标识和目标账户的命令。
* @param {string} device_type - 0：分区设备 1：寻呼台设备
*
* @description
* 发送的消息格式如下：
* {
* command: "get_user_zone", 获取用户的分区：get_user_zone
* dest_account: "目标账户" // 目标账户名称（自己或子用户）
* }
*/
export const getDeviceInfo = (type) => {
return new Promise(async (resolve, reject) => {
// 如果 WebSocket 未连接，直接拒绝
if (!socket || socket.readyState !== WebSocket.OPEN) {
await initWebSocket()
console.warn('WebSocket 未连')
return reject(new Error('WebSocket 未连接'))
}
const Message = {
uuid: '登录返回的uuid',
command: 'get_device_info',
device_type: type, // 0：分区设备 1：寻呼台设备
all_zone: true, // 是否请求全部分区
page: 1
}
try {
socket.send(JSON.stringify(Message))
resolve()
} catch (error) {
return reject(new Error('发送消息失败: ' + error.message))
}
})
}
5. 请求示例
<script setup>
import { onMounted, onBeforeUnmount } from 'vue'
import { initWebSocket, closeWebSocket, getDeviceInfo } from '@/utils/WebSocket'
import { subscribe } from '@/utils/eventBus'
// 订阅消息
subscribe((ev) => {
if (ev.command == 'get_device_info') {
// 这里处理订阅的消息
}
})
onMounted(async () => {
await initWebSocket() // 初始化websocket
await getDeviceInfo('3') // 获取设备信息
})
onBeforeUnmount(() => {
closeWebSocket() // 关闭websocket
})
</script>
<template>
<div></div>
</template>
<style lang="scss" scoped></style>
6. 完整代码
// websocketService.js
// websoket链接（用于IP广播）
import { notifyMessage } from '@/utils/eventBus' // 引入消息总线
let socket = null
let connectionStatus = 'disconnected'
let connectPromise = null // 核心：保存连接的 Promise，用于共享连接状态
// 可配置的最大重试次数和重连间隔
const MAX_RECONNECT_ATTEMPTS = 5
const RECONNECT_INTERVAL = 3000 // 3秒
let reconnectAttempts = 0
let shouldReconnect = false
let onMessageCallback = null
let timer = null
/**
* 初始化 WebSocket 连接（Promise 版本）
*
* @param {string} wsUrl - WebSocket 服务器地址
* @param {function} onMessage - 可选的消息回调函数
* @param {number} timeout - 可选：连接超时时间（毫秒），默认 10 秒
* @returns {Promise<WebSocket>} - 成功时返回 socket 实例
*/
export const initWebSocket = (wsUrl = 'ws://') => {
// 如果已经有连接或正在连接，直接返回同一个 Promise
if (connectPromise) {
return connectPromise
}
// 创建新的连接 Promise
connectPromise = new Promise((resolve, reject) => {
// 如果已经连接，直接 resolve
if (socket && socket.readyState === WebSocket.OPEN) {
console.log('WebSocket 已连接，跳过初始化')
return resolve()
}
// 正在连接或手动关闭后不再自动重连，则拒绝
if (socket && socket.readyState === WebSocket.CONNECTING) {
console.log('WebSocket 正在连接中...')
return
}
shouldReconnect = true
reconnectAttempts = 0
socket = new WebSocket(wsUrl)
connectionStatus = 'connecting'
socket.onopen = () => {
console.log('WebSocket 连接成功')
connectionStatus = 'connected'
sessionStorage.removeItem('storage-token')
// 连接成功后尝试登录
userLogin('admin', 'admin')
.then(() => {
console.log('自动登录成功')
startHeartbeat() // 登录成功后开始心跳
resolve() // 登录成功才认为初始化完成
})
.catch((err) => {
console.error('自动登录失败:', err)
reject(new Error('登录失败'))
})
}
socket.onmessage = (event) => {
let data
try {
data = JSON.parse(event.data)
if (data.command == 'heartbeat' && data.result != 0) closeWebSocket()
} catch (e) {
console.error('无法解析消息:', event.data)
return
}
// 处理登录响应
if (data.command === 'user_login') {
handleLoginResponse(data, resolve, reject)
return
}
// 广播其他消息
notifyMessage(data)
}
socket.onerror = (error) => {
console.error('WebSocket 错误:', error)
connectionStatus = 'error'
}
socket.onclose = () => {
console.log('WebSocket 连接关闭')
connectionStatus = 'disconnected'
clearInterval(timer)
if (shouldReconnect && reconnectAttempts < MAX_RECONNECT_ATTEMPTS) {
reconnectAttempts++
console.log(`尝试重连... (${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})`)
setTimeout(() => {
connectPromise = null // 允许重新连接
initWebSocket(wsUrl).catch(() => {})
}, RECONNECT_INTERVAL)
} else {
console.warn('停止重连')
}
}
})
return connectPromise
}
// 2. 发送消息函数
export const sendMessage = async (message) => {
if (!message) {
console.warn('无法发送空消息')
return false
}
try {
// 确保连接已建立
await initWebSocket()
} catch (error) {
console.warn('连接失败，无法发送消息:', error.message)
return false
}
if (socket.readyState === WebSocket.OPEN) {
try {
socket.send(JSON.stringify(message))
return true
} catch (error) {
console.error('发送消息失败:', error)
return false
}
} else {
console.warn('WebSocket 未处于 OPEN 状态，无法发送')
return false
}
}
// 3. 关闭连接函数
export const closeWebSocket = () => {
shouldReconnect = false
if (socket) {
socket.close()
}
if (timer) clearInterval(timer)
connectPromise = null // 允许下次重新连接
sessionStorage.removeItem('storage-token')
}
// 登录响应处理
let loginResolve = null
let loginReject = null
function handleLoginResponse(data, resolve, reject) {
if (data.result === 0) {
try {
sessionStorage.setItem('storage-name', data.user_name || '')
sessionStorage.setItem('storage-password', data.password || '')
sessionStorage.setItem('storage-token', data.uuid || '')
sessionStorage.setItem('storage-userType', data.user_type || '')
} catch (err) {
console.error('存储登录信息失败:', err)
}
if (loginResolve) loginResolve(data)
if (resolve) resolve()
} else {
const error = new Error(data.msg || '登录失败')
if (loginReject) loginReject(error)
if (reject) reject(error)
}
loginResolve = null
loginReject = null
}
/**
* 用户登录函数
*
* 该函数用于处理用户登录请求
* @param {string} account - 用户的登录账户名（可以是用户名、邮箱或手机号等）。
* @param {string} password - 用户的登录密码。建议在调用此函数前对密码进行加密处理，避免明文传输。
* @description
* 发送的消息格式如下：
* {
* command: "user_login ", // 指定操作为用户登录（注意：末尾有多余空格）
* account: "用户账户",
* password: "用户密码"
* }
*/
export const userLogin = (account, password) => {
return new Promise((resolve, reject) => {
if (!socket || socket.readyState !== WebSocket.OPEN) {
return reject(new Error('WebSocket 未连接'))
}
const token = sessionStorage.getItem('storage-token')
if (token) {
return resolve({ status: 'success', msg: 'already logged in' })
}
loginResolve = resolve
loginReject = reject
const loginMessage = {
command: 'user_login',
account: '',
password: ''
}
try {
socket.send(JSON.stringify(loginMessage))
} catch (error) {
reject(new Error('发送登录消息失败: ' + error.message))
}
})
}
/**
* 心跳检测
*/
const startHeartbeat = () => {
clearInterval(timer)
timer = setInterval(() => {
if (socket.readyState === WebSocket.OPEN) {
const heartbeatMsg = {
uuid: sessionStorage.getItem('storage-token'),
command: 'heartbeat'
}
try {
socket.send(JSON.stringify(heartbeatMsg))
} catch (e) {
console.error('心跳发送失败')
}
}
}, 60000)
}
/**
* 4.2 获取设备信息
*
* 该函数用于向指定设备发送指令，以获取与指定账户关联的区域（zone）信息。
* 它通过调用 sendMessage 函数发送一个包含设备唯一标识和目标账户的命令。
* @param {string} device_type - 0：分区设备 1：寻呼台设备
*
* @description
* 发送的消息格式如下：
* {
* command: "get_user_zone", 获取用户的分区：get_user_zone
* dest_account: "目标账户" // 目标账户名称（自己或子用户）
* }
*/
export const getDeviceInfo = (type) => {
return new Promise(async (resolve, reject) => {
// 如果 WebSocket 未连接，直接拒绝
if (!socket || socket.readyState !== WebSocket.OPEN) {
await initWebSocket()
console.warn('WebSocket 未连')
return reject(new Error('WebSocket 未连接'))
}
// 发送登录消息（注意：原代码 command 末尾有空格，按原逻辑保留）
const Message = {
uuid: sessionStorage.getItem('storage-token'),
command: 'get_device_info',
device_type: type, // 0：分区设备 1：寻呼台设备
all_zone: true, // 是否请求全部分区
page: 1
// zone_mac:'' // 指定分区的mac地址：all_zone=0时此字段有效
}
try {
socket.send(JSON.stringify(Message))
resolve()
} catch (error) {
return reject(new Error('发送消息失败: ' + error.message))
}
})
}
