1. 移动socket客户端类、新增socket/index.js文件（初始化） 2. socketMessage枚举类的完善（增加对每个类型的英文描...

hcong

2024-11-26 d667fce729c223e885930bb8edb528ab501b8d6a

1. 移动socket客户端类、新增socket/index.js文件（初始化） 2. socketMessage枚举类的完善（增加对每个类型的英文描述字段以及对类型筛选和判断是否为心跳消息的工具方法） 3. 在FYBgTaskCard.vue监听并处理如何刷后台任务状态逻辑

 src/App.vue

@@ -40,8 +40,6 @@

<script>
import zhCn from 'element-plus/dist/locale/zh-cn.mjs';
import { FYWebSocket } from '@/socket/index.js'
import eventBus from '@/socket/eventBus.js';
export default {
  data() {
    return {
@@ -58,25 +56,6 @@
    navPage(titles) {
      this.navTitles = titles;
    },
    // è¿žæŽ¥websocket
    connectWebSocket() {
      this.socket = new FYWebSocket()
      this.socket.init({
        time: 4 * 1000,
        timeout: 2 * 1000,
        reconnect: 3 * 1000
      }, true)
    },
    // æ–­çº¿é‡è¿žwebsocket
    startReconnectWebSocket() {
      eventBus.register('reconnect', () => {
        this.connectWebSocket()
      })
    }
  },
  created() {
    this.connectWebSocket()
    this.startReconnectWebSocket()
  }
};
</script>

 src/components/bg-task/FYBgTaskCard.vue

@@ -46,6 +46,7 @@
import { useFetchData } from '@/composables/fetchData';
import bgtaskApi from '@/api/fysp/bgtaskApi';
import { enumBgTask, BG_TASK_TYPE, BG_TASK_STATUS } from '@/enum/bgTask';
import eventBus from '@/socket/eventBus.js';

export default {
  setup() {
@@ -77,6 +78,32 @@
    }
  },
  methods: {
    registerBgTaskMessage() {
      eventBus.register('background_task', (data) => {
        this.refreshTaskById(data)
      })
    },
    /**
     * åˆ·æ–°ä¸€ä¸ªä»»åŠ¡é€šè¿‡id，如果是新的任务则添加到任务列表taskList中
     * @param data 
     */
    refreshTaskById(data) {
      if (!data || data == {}) {
        return;
      }
      let isNewTask = true
      for (let index = 0; index < this.taskList.length; index++) {
          const task = this.taskList[index];
          if (task.id == data.id) {
            this.taskList[index] = data
            isNewTask = false
            break
          }
        }
        if (isNewTask) {
          this.taskList.push(data)
        }
    },
    addTask() {},
    newTestTask() {
      this.fetchData((page, pageSize) => {
@@ -151,7 +178,11 @@
    },
    gotoResult(index) {}
  },
  created() {
    this.registerBgTaskMessage()
  },
  mounted() {
    bgtaskApi.newTestTask(1)
    this.fetchTask();
    // setInterval(() => {
    //   this.fetchTask();

 src/enum/socketMessage.js

@@ -1,6 +1,17 @@
const messageTypes = [
  { id: 0, label: '心跳检测', value: '0' },
  { id: 1, label: '后台任务', value: '1' },
  { id: 2, label: '业务日志', value: '2' }
  { id: 0, value: '0', label: '心跳机制', name: 'heartbeat' },
  { id: 1, value: '1', label: '后台任务', name: 'background_task' },
  { id: 2, value: '2', label: '业务日志', name: 'business_log' }
];
export { messageTypes };
/**
 * æ ¹æ®socket消息类型字段得到这个类型对象
 * @param {*} value socket消息类型
 * @returns å¯¹åº”的消息类型对象
 */
function getByValue(value) {
  return messageTypes.find((v) => v.value == value);
}
function isHeartbeatMessageByType(type) {
  return type == '0';
}
export { messageTypes, getByValue, isHeartbeatMessageByType };

 src/main.js

@@ -5,6 +5,7 @@
import { router } from './router';
import App from './App.vue';
import timeUtil from './utils/time-util';
import { initSocketClient } from '@/socket/index.js'

// import 'element-plus/dist/index.css';
import './assets/main.css';
@@ -19,6 +20,9 @@
import isSameOrAfter from 'dayjs/plugin/isSameOrAfter';
import isSameOrBefore from 'dayjs/plugin/isSameOrBefore';

// socket客户端
initSocketClient()

dayjs.extend(isSameOrAfter);
dayjs.extend(isSameOrBefore);


 src/socket/FYWebSocket.js

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,134 @@
import eventBus from './eventBus.js';
import {
  messageTypes,
  getByValue,
  isHeartbeatMessageByType
} from '@/enum/socketMessage.js';
import { encodeMessage, decodeMessage } from './socketMessage.js';
import { $socket_base_url } from '@/api/index.js';
class FYWebSocket extends WebSocket {
  constructor() {
    super($socket_base_url);
    return this;
  }
  /**
   * heartBeatConfig å¿ƒè·³è¿žæŽ¥å‚æ•°
   *    time: å¿ƒè·³æ—¶é—´é—´éš”
   *    timeout: å¿ƒè·³è¶…时间隔
   *    reconnect: æ–­çº¿é‡è¿žæ—¶é—´é—´éš”
   * isReconnect æ˜¯å¦æ–­çº¿é‡è¿ž
   */
  init(heartBeatConfig, isReconnect) {
    this.onopen = this.openHandler; // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
    this.onclose = this.closeHandler; // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
    this.onmessage = this.messageHandler; // æ”¶åˆ°æœåŠ¡å™¨æ•°æ®åŽçš„回调函数
    this.onerror = this.errorHandler; // è¿žæŽ¥å‘生错误的回调方法
    this.heartBeatConfig = heartBeatConfig; // å¿ƒè·³è¿žæŽ¥é…ç½®å‚æ•°
    this.isReconnect = isReconnect; // è®°å½•æ˜¯å¦æ–­çº¿é‡è¿ž
    this.reconnectTimer = null; // è®°å½•æ–­çº¿é‡è¿žçš„时间器
    this.startHeartBeatTimer = null; // è®°å½•å¿ƒè·³æ—¶é—´å™¨
    this.webSocketState = false; // è®°å½•socket连接状态 true为已连接
  }
  // èŽ·å–消息
  getMessage({ data }) {
    return decodeMessage(data);
  }
  // å‘送消息
  sendMessage(type, data) {
    return this.send(encodeMessage(type, data));
  }
  // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
  openHandler() {
    console.log('====onopen è¿žæŽ¥æˆåŠŸ====');
    // socket状态设置为连接，做为后面的断线重连的拦截器
    this.webSocketState = true;
    // åˆ¤æ–­æ˜¯å¦å¯åŠ¨å¿ƒè·³æœºåˆ¶
    if (this.heartBeatConfig && this.heartBeatConfig.time) {
      this.startHeartBeat(this.heartBeatConfig.time);
    }
  }
  // æ”¶åˆ°æœåŠ¡å™¨æ•°æ®åŽçš„回调函数
  messageHandler(res) {
    const webSocketMsg = this.getMessage(res);
    if (!(webSocketMsg && webSocketMsg != null && webSocketMsg != {})) {
      return;
    }
    console.log("webSocketMsg", webSocketMsg);
    
    const type = webSocketMsg.type;
    const data = webSocketMsg.data;
    let typeObj = getByValue(type);
    console.log("typeObj", typeObj);
    
    if (isHeartbeatMessageByType(type)) {
      // å°†è¿žæŽ¥çŠ¶æ€æ›´æ–°ä¸ºåœ¨çº¿
      this.webSocketState = true;
      eventBus.emit(typeObj.name, data);
      console.log('====onmessage å¿ƒè·³æ£€æµ‹====', data);
    } else {
      // å‘送事件
      eventBus.emit(typeObj.name, data);
      console.log(`====onmessage ${typeObj.label}====`, data);
    }
  }
  // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
  closeHandler() {
    console.log('====onclose websocket关闭连接====');
    // è®¾ç½®socket状态为断线
    this.webSocketState = false;
    // åœ¨æ–­å¼€è¿žæŽ¥æ—¶ æ¸…除心跳时间器和 æ–­å¼€é‡è¿žæ—¶é—´å™¨
    this.startHeartBeatTimer && clearTimeout(this.startHeartBeatTimer);
    this.reconnectTimer && clearTimeout(this.reconnectTimer);
    this.reconnectWebSocket();
  }
  errorHandler() {
    console.log('====onerror websocket连接出错====');
    // è®¾ç½®socket状态为断线
    this.webSocketState = false;
    // é‡æ–°è¿žæŽ¥
    this.reconnectWebSocket();
  }

  // å¿ƒè·³åˆå§‹åŒ–方法 time：心跳间隔
  startHeartBeat(time) {
    this.startHeartBeatTimer = setTimeout(() => {
      // å®¢æˆ·ç«¯æ¯éš”一段时间向服务端发送一个心跳消息
      this.sendMessage(messageTypes[0].value, Date.now());
      this.waitingServer();
    }, time);
  }
  //在客户端发送消息之后，延时等待服务器响应,通过webSocketState判断是否连线成功
  waitingServer() {
    this.webSocketState = false;
    setTimeout(() => {
      // è¿žçº¿æˆåŠŸçŠ¶æ€ä¸‹ ç»§ç»­å¿ƒè·³æ£€æµ‹
      if (this.webSocketState) {
        this.startHeartBeat(this.heartBeatConfig.time);
        return;
      }
      console.log('心跳无响应, å·²ç»å’ŒæœåŠ¡ç«¯æ–­çº¿');
      // é‡æ–°è¿žæŽ¥æ—¶ï¼Œè®°å¾—要先关闭当前连接
      try {
        this.close();
      } catch (error) {
        console.log('当前连接已经关闭');
      }
      // // é‡æ–°è¿žæŽ¥
      // this.reconnectWebSocket()
    }, this.heartBeatConfig.timeout);
  }

  // é‡æ–°è¿žæŽ¥
  reconnectWebSocket() {
    // åˆ¤æ–­æ˜¯å¦æ˜¯é‡æ–°è¿žæŽ¥çŠ¶æ€(即被动状态断线)，如果是主动断线的不需要重新连接
    if (!this.isReconnect) {
      return;
    }
    // æ ¹æ®ä¼ å…¥çš„断线重连时间间隔 å»¶æ—¶è¿žæŽ¥
    this.reconnectTimer = setTimeout(() => {
      // è§¦å‘重新连接事件
      eventBus.emit('reconnect');
    }, this.heartBeatConfig.reconnect);
  }
}
export { FYWebSocket };

 src/socket/index.js

@@ -1,134 +1,31 @@
import eventBus from './eventBus.js';
import { messageTypes } from '@/enum/socketMessage.js';
import { encodeMessage, decodeMessage } from './socketMessage.js';
import { $socket_base_url } from '@/api/index.js';
class FYWebSocket extends WebSocket {
  constructor() {
    super($socket_base_url);
    return this;
  }
  /**
   * heartBeatConfig å¿ƒè·³è¿žæŽ¥å‚æ•°
   *    time: å¿ƒè·³æ—¶é—´é—´éš”
   *    timeout: å¿ƒè·³è¶…时间隔
   *    reconnect: æ–­çº¿é‡è¿žæ—¶é—´é—´éš”
   * isReconnect æ˜¯å¦æ–­çº¿é‡è¿ž
   */
  init(heartBeatConfig, isReconnect) {
    this.onopen = this.openHandler; // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
    this.onclose = this.closeHandler; // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
    this.onmessage = this.messageHandler; // æ”¶åˆ°æœåŠ¡å™¨æ•°æ®åŽçš„回调函数
    this.onerror = this.errorHandler; // è¿žæŽ¥å‘生错误的回调方法
    this.heartBeatConfig = heartBeatConfig; // å¿ƒè·³è¿žæŽ¥é…ç½®å‚æ•°
    this.isReconnect = isReconnect; // è®°å½•æ˜¯å¦æ–­çº¿é‡è¿ž
    this.reconnectTimer = null; // è®°å½•æ–­çº¿é‡è¿žçš„时间器
    this.startHeartBeatTimer = null; // è®°å½•å¿ƒè·³æ—¶é—´å™¨
    this.webSocketState = false; // è®°å½•socket连接状态 true为已连接
  }
  // èŽ·å–消息
  getMessage({ data }) {
    return decodeMessage(data);
  }
  // å‘送消息
  sendMessage(type, data) {
    return this.send(encodeMessage(type, data));
  }
  // è¿žæŽ¥æˆåŠŸåŽçš„回调函数
  openHandler() {
    console.log('====onopen è¿žæŽ¥æˆåŠŸ====');
    // socket状态设置为连接，做为后面的断线重连的拦截器
    this.webSocketState = true;
    // åˆ¤æ–­æ˜¯å¦å¯åŠ¨å¿ƒè·³æœºåˆ¶
    if (this.heartBeatConfig && this.heartBeatConfig.time) {
      this.startHeartBeat(this.heartBeatConfig.time);
    }
  }
  // æ”¶åˆ°æœåŠ¡å™¨æ•°æ®åŽçš„回调函数
  messageHandler(res) {
    const webSocketMsg = this.getMessage(res);
    if (!(webSocketMsg && webSocketMsg != null && webSocketMsg != {})) {
      return;
    }
    const type = webSocketMsg.type;
    const data = webSocketMsg.data
    
    switch (type) {
      case '1': // åŽå°ä»»åŠ¡
        // å‘送事件
        eventBus.emit('1', data);
        console.log('====onmessage åŽå°ä»»åŠ¡====', data);
        break;
      case '2': // ä¸šåŠ¡æ—¥å¿—
        // å‘送事件
        eventBus.emit('2', data);
        console.log('====onmessage ä¸šåŠ¡æ—¥å¿—====', data);
        break;
      case '0': // å¿ƒè·³æ£€æµ‹
        // å°†è¿žæŽ¥çŠ¶æ€æ›´æ–°ä¸ºåœ¨çº¿
        this.webSocketState = true;
        eventBus.emit('0', data);
        console.log('====onmessage å¿ƒè·³æ£€æµ‹====', data);
        break;
    }
  }
  // è¿žæŽ¥å…³é—­åŽçš„回调 å‡½æ•°
  closeHandler() {
    console.log('====onclose websocket关闭连接====');
    // è®¾ç½®socket状态为断线
    this.webSocketState = false;
    // åœ¨æ–­å¼€è¿žæŽ¥æ—¶ æ¸…除心跳时间器和 æ–­å¼€é‡è¿žæ—¶é—´å™¨
    this.startHeartBeatTimer && clearTimeout(this.startHeartBeatTimer);
    this.reconnectTimer && clearTimeout(this.reconnectTimer);
    this.reconnectWebSocket();
  }
  errorHandler() {
    console.log('====onerror websocket连接出错====');
    // è®¾ç½®socket状态为断线
    this.webSocketState = false;
    // é‡æ–°è¿žæŽ¥
    this.reconnectWebSocket();
  }

  // å¿ƒè·³åˆå§‹åŒ–方法 time：心跳间隔
  startHeartBeat(time) {
    this.startHeartBeatTimer = setTimeout(() => {
      // å®¢æˆ·ç«¯æ¯éš”一段时间向服务端发送一个心跳消息
      this.sendMessage(messageTypes[0].value, Date.now());
      this.waitingServer();
    }, time);
  }
  //在客户端发送消息之后，延时等待服务器响应,通过webSocketState判断是否连线成功
  waitingServer() {
    this.webSocketState = false;
    setTimeout(() => {
      // è¿žçº¿æˆåŠŸçŠ¶æ€ä¸‹ ç»§ç»­å¿ƒè·³æ£€æµ‹
      if (this.webSocketState) {
        this.startHeartBeat(this.heartBeatConfig.time);
        return;
      }
      console.log('心跳无响应, å·²ç»å’ŒæœåŠ¡ç«¯æ–­çº¿');
      // é‡æ–°è¿žæŽ¥æ—¶ï¼Œè®°å¾—要先关闭当前连接
      try {
        this.close();
      } catch (error) {
        console.log('当前连接已经关闭');
      }
      // // é‡æ–°è¿žæŽ¥
      // this.reconnectWebSocket()
    }, this.heartBeatConfig.timeout);
  }
import { FYWebSocket } from '@/socket/FYWebSocket.js'
import eventBus from '@/socket/eventBus.js';

  // é‡æ–°è¿žæŽ¥
  reconnectWebSocket() {
    // åˆ¤æ–­æ˜¯å¦æ˜¯é‡æ–°è¿žæŽ¥çŠ¶æ€(即被动状态断线)，如果是主动断线的不需要重新连接
    if (!this.isReconnect) {
      return;
    }
    // æ ¹æ®ä¼ å…¥çš„断线重连时间间隔 å»¶æ—¶è¿žæŽ¥
    this.reconnectTimer = setTimeout(() => {
      // è§¦å‘重新连接事件
      eventBus.emit('reconnect');
    }, this.heartBeatConfig.reconnect);
let socket = null;
// è¿žæŽ¥websocket
function connectWebSocket() {
  if (socket == null) {
    socket = new FYWebSocket()
  }
  socket.init({
    time: 4 * 1000,
    timeout: 2 * 1000,
    reconnect: 3 * 1000
  }, true)
  return socket;
}
export { FYWebSocket };
// æ–­çº¿é‡è¿žwebsocket
function startReconnectWebSocket() {
  eventBus.register('reconnect', () => {
    connectWebSocket()
  })
}
/**
 * åˆå§‹åŒ–socket客户端
 */
function initSocketClient() {
  connectWebSocket();
  startReconnectWebSocket();
}
export { initSocketClient }

 src/socket/socketMessage.js

@@ -7,7 +7,7 @@
// ç»“束符号
const endStr = '%%';
// æ ¡éªŒæ ¼å¼
function verificationMessahe(message) {
function verificationMessage(message) {
  if (!message || message == '') {
    return false;
  }
@@ -28,7 +28,7 @@
 * @returns 
 */
function decodeMessage(message) {
  if (!verificationMessahe(message)) {
  if (!verificationMessage(message)) {
    return;
  }
  const parts = message.slice(startStr.length, -endStr.length).split(splitStr);
@@ -49,4 +49,4 @@
  return `${startStr}${type}${splitStr}${JSON.stringify(data)}${endStr}`;
}

export { verificationMessahe, decodeMessage, encodeMessage, messageTypes };
export { verificationMessage, decodeMessage, encodeMessage };