当前位置：首页 > article >正文

WebSocket实现消息实时推送

article 2024/11/7 4:11:47

文章目录

- websocket介绍
- - 特点
  - 工作原理
- 用websocket实现实时推送
- - 引入依赖
  - WebSocket 函数定义
  - 变量声明
  - 初始化 WebSocket 连接
  - WebSocket 连接的初始化和事件处理
  - 连接打开事件
  - 接收消息处理
  - 连接关闭和重连机制
  - 心跳机制
  - 使用 WebSocket
  - 代码完整显示

websocket介绍

WebSocket 是一种网络通信协议，旨在实现客户端和服务器之间的双向通信。它允许在单个 TCP 连接上进行全双工（即同时进行发送和接收）通信。WebSocket 特别适用于需要实时更新的应用，比如在线游戏、聊天应用、实时数据流等。

特点

双向通信：与传统的 HTTP 请求-响应模型不同，WebSocket 允许客户端和服务器同时发送和接收数据，这意味着一方可以主动向另一方发送消息，而不需要等待请求。

持久连接：WebSocket 建立的连接是持久的，客户端和服务器在一次连接后可以持续交换消息，而不需要频繁建立和关闭连接，从而减少了网络延迟和开销。

低延迟：由于使用了持久连接，WebSocket 可以减少消息传输中的延迟，使实时应用的响应速度更快。

轻量级：WebSocket 协议在数据传输中相对轻量，不需要像 HTTP 那样包含大量的头信息，数据包开销更小。

工作原理

握手：首先，客户端通过发送 HTTP 请求向服务器发起 WebSocket 连接。请求中包含一些特定的头信息，表明希望升级到 WebSocket 协议。

建立连接：服务器收到请求后，如果支持 WebSocket，将返回一个响应，确认升级连接。此时，HTTP 连接转变为 WebSocket 连接。

数据传输：连接建立后，双方可以自由地发送和接收消息。消息格式可以是文本（如 JSON）或二进制数据。

关闭连接：一方可以发送关闭帧，另一方收到后也会关闭连接，整个通信过程结束。

用websocket实现实时推送

这个封装提供了一个基础的 WebSocket 功能，可以支持实时消息推送。通过增加事件监听的管理、错误处理、灵活的心跳机制等功能，可以使这个封装更加健壮和灵活。

引入依赖

import { ElMessage, getCache } from "@/utils";

ElMessage: 用于显示消息提示。
getCache: 用于获取缓存数据，这里用来获取用户ID。

WebSocket 函数定义

function webSocket(params: string) {
  ...
  return {
    onMessageFns,
    init,
  };
}

webSocket 函数接收一个参数 params（通常是用户ID），用于建立特定的 WebSocket 连接。

变量声明

  let urlParams: string = params;
  const isReconnect = ref(true);
  let reconnection: any;
  let ws: WebSocket | null = null;
  let websocketConnectedCount = 0;
  let serverTimeoutObj: any = null;
  const hearbeatInterval = 300000; // 心跳间隔

is_reconnect: 用于标识是否可以重连。
ws: 存储 WebSocket 实例。
websocketConnectdCount : 记录连接失败次数。
serverTimeoutObj : 用于管理心跳检测的定时器。

初始化 WebSocket 连接

const init = () => {
  if (!("WebSocket" in window)) {
    ElMessage({
      message: "抱歉，浏览器不支持Websocket！",
      type: "warning",
      duration: 1000,
    });
    return;
  }
  try {
    initWebSocket();
  } catch (e) {
    console.log("尝试创建连接失败");
    reConnect();
  }
};

首先检查浏览器是否支持 WebSocket。如果支持，则尝试初始化连接；如果失败，则调用 reConnect 进行重连。

WebSocket 连接的初始化和事件处理

 function initWebSocket() {
    const baseUrl = import.meta.env.VITE_BaseUrl?.slice(7);
    const url = `ws://${baseUrl}/websocket/${urlParams}`;
    ws = new WebSocket(url);

    ws.onopen = (e: Event) => {
      websocketOpen(e);
    };

    ws.onmessage = (e: MessageEvent) => {
      websocketOnMessage(e);
    };

    ws.onerror = () => {
      console.log("WebSocket连接发生错误");
      isReconnect.value = false;
      websocketConnectedCount++;
      if (websocketConnectedCount <= 5) {
        reConnect();
      }
    };

    ws.onclose = (e: CloseEvent) => {
      websocketClose(e);
    };
  }

创建 WebSocket 实例，并设置各类事件处理函数（打开、接收消息、错误、关闭）。

连接打开事件

function websocketOpen(e: Event) {
    console.log("连接成功");
    reset();
    start();
    const data = { sendType: "HEALTH" };
    ws?.send(JSON.stringify(data));
  }
}

连接成功时，重置心跳并发送健康检查消息。

接收消息处理

const onMessageFns = new Set<(e: MessageEvent) => void>();
  function websocketOnMessage(e: MessageEvent) {
    onMessageFns.forEach((callback) => callback(e));
    reset();
    start();
    return e.data;
  }

接收到消息时，执行所有注册的回调函数，并重置心跳。

连接关闭和重连机制

function websocketclose(e: any) {
  console.log(e);
  is_reconnect.value = false;
  console.log("connection closed (" + e.code + ")");
}

let reConnect = () => {
  console.log("尝试重新连接");
  if (is_reconnect) return; // 如果已经连上就不再重连
  reconnection && clearTimeout(reconnection);
  reconnection = setTimeout(function () {
    init();
  }, 5000);
};

连接关闭时设置标识，并在指定时间后尝试重连。

心跳机制

 const reset = () => {
    clearTimeout(serverTimeoutObj);
  };

  const start = () => {
    serverTimeoutObj = setInterval(() => {
      if (ws?.readyState === WebSocket.OPEN) {
        console.log("连接状态，发送消息保持连接");
        const data = { sendType: "HEALTH" };
        ws?.send(JSON.stringify(data));
        reset();
      } else {
        console.log("断开连接， 尝试重连");
        webSocket(urlParams);
      }
    }, hearbeatInterval);
  };

定期发送健康检查消息，保持连接活跃。

使用 WebSocket

import webSocket from "@/utils/websocket";

const userId = getCache("userId");
const ws = webSocket(userId)!;

onMounted(async () => {
  const addMessage = () => {
    ws.onMessageFns.add((value: MessageEvent) => {
      try {
        if (value.data !== "来自后台的反馈：连接成功" && value.data !== "SUCCESS") {
          const parsedData = JSON.parse(value.data);
          console.log("获取到的信息", parsedData);
        }
      } catch (error) {
        console.error("消息解析错误:", error);
      }
    });
  };
  addMessage();
});

在组件挂载时，获取用户ID并创建 WebSocket 实例。添加消息接收处理，过滤特定消息并处理 JSON 数据。

代码完整显示

import { ElMessage, getCache } from "@/utils";

function webSocket(params: string) {
  let urlParams: string = params;
  const isReconnect = ref(true);
  let reconnection: any;
  let ws: WebSocket | null = null;
  let websocketConnectedCount = 0;
  let serverTimeoutObj: any = null;
  const hearbeatInterval = 300000; // 心跳间隔

  const init = () => {
    if (!("WebSocket" in window)) {
      ElMessage({
        message: "抱歉，浏览器不支持WebSocket！",
        type: "warning",
        duration: 1000,
      });
      return;
    }
    try {
      initWebSocket();
    } catch (e) {
      console.log("尝试创建连接失败");
      reConnect();
    }
  };

  function initWebSocket() {
    const baseUrl = import.meta.env.VITE_BaseUrl?.slice(7);
    const url = `ws://${baseUrl}/websocket/${urlParams}`;
    ws = new WebSocket(url);

    ws.onopen = (e: Event) => {
      websocketOpen(e);
    };

    ws.onmessage = (e: MessageEvent) => {
      websocketOnMessage(e);
    };

    ws.onerror = () => {
      console.log("WebSocket连接发生错误");
      isReconnect.value = false;
      websocketConnectedCount++;
      if (websocketConnectedCount <= 5) {
        reConnect();
      }
    };

    ws.onclose = (e: CloseEvent) => {
      websocketClose(e);
    };
  }

  function websocketOpen(e: Event) {
    console.log("连接成功");
    reset();
    start();
    const data = { sendType: "HEALTH" };
    ws?.send(JSON.stringify(data));
  }

  const onMessageFns = new Set<(e: MessageEvent) => void>();
  function websocketOnMessage(e: MessageEvent) {
    onMessageFns.forEach((callback) => callback(e));
    reset();
    start();
    return e.data;
  }

  function websocketClose(e: CloseEvent) {
    console.log("connection closed (" + e.code + ")");
    isReconnect.value = false;
  }

  const reConnect = () => {
    console.log("尝试重新连接");
    if (isReconnect.value) return; // 如果已经连上就不再重连
    reconnection && clearTimeout(reconnection);
    reconnection = setTimeout(init, 5000);
  };

  const reset = () => {
    clearTimeout(serverTimeoutObj);
  };

  const start = () => {
    serverTimeoutObj = setInterval(() => {
      if (ws?.readyState === WebSocket.OPEN) {
        console.log("连接状态，发送消息保持连接");
        const data = { sendType: "HEALTH" };
        ws?.send(JSON.stringify(data));
        reset();
      } else {
        console.log("断开连接， 尝试重连");
        webSocket(urlParams);
      }
    }, hearbeatInterval);
  };

  return {
    onMessageFns,
    init,
  };
}

// 使用 WebSocket
import webSocket from "@/utils/websocket";

const userId = getCache("userId");
const ws = webSocket(userId)!;

onMounted(async () => {
  const addMessage = () => {
    ws.onMessageFns.add((value: MessageEvent) => {
      try {
        if (value.data !== "来自后台的反馈：连接成功" && value.data !== "SUCCESS") {
          const parsedData = JSON.parse(value.data);
          console.log("获取到的信息", parsedData);
        }
      } catch (error) {
        console.error("消息解析错误:", error);
      }
    });
  };
  addMessage();
});