• vue3+ts+java使用WebSocket传输数据

    一、环境

    系统：win11

    IDE：vscode

    框架：electron22.0.0+vite2+vue3+typescript4.8.4+springboot2.2.5+jdk1.8

    二、websocket介绍

    2.1 由来

            WebSocket未出现之前，浏览器和服务器之间的通信是通过Web的poll技术进行通信，就是浏览器不停的对服务器主动发动请求，每发起一次新的HTTP请求，就是开启一次新的TCP链接，HTTP协议本身就是一种开销非常大的协议，所以这种方式效率低下。于是就出现了WebSocket协议。

            下面是采用poll方式的代码示例：

    1. setInterval(() => {
    2.   // 查询注册机列表
    3.   getRegisterInfo().then(res => {
    4.      isHost.value = store.state.onHost;   
    5.   }).catch(err => {
    6.      console.log('getComputerList err:', err);
    7.   });
    8. }, 1000);

    为了页面及时更新，会像服务器产生大量的请求，造成资源浪费。

    2.2 WebSocket通信过程

            WebSocket是一种完全不同于HTTP的协议，但是它需要通过HTTP协议的GET请求，将HTTP协议升级为WebSocket协议。升级的过程被称为握手(handshake)。当浏览器和服务器握手成功后，则可以开始根据WebSocket定义的通信帧格式开始通信了。WebSocket协议的通信帧也分为控制数据帧和普通数据帧，前者用于控制WebSocket链接状态，后者用于承载数据。

            握手过程就是将HTTP协议升级为WebSocket协议的过程。在HTTP的GET请求头部添加信息如下：

    Upgrade: websocket      #规定必需的字段，其值必需为 websocket , 如果不是则握手失败；
    Connection: Upgrade  #规定必需的字段，值必需为 Upgrade , 如果不是则握手失败；
    Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==    #必需字段，一个随机的字符串；
    Sec-WebSocket-Protocol: chat, superchat   #可选字段，可以用于标识应用层的协议；
    Sec-WebSocket-Version: 13  #必需字段，代表了 WebSocket 协议版本，值必需是 13 , 否则
    握手失败；

    当服务器端，成功验证了以上信息后，则会返回一个形如以下信息的响应:

    HTTP/1.1 101 Switching Protocols   #101代表握手成功的状态码
    Upgrade: websocket  #规定必需的字段，其值必需为 websocket , 如果不是则握手失败；
    Connection: Upgrade #规定必需的字段，值必需为 Upgrade , 如果不是则握手失败；
    Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=  #规定必需的字段，该字段的值是通过固定字符串 258EAFA5-E914-47DA-95CA-C5AB0DC85B11 加上请求中 Sec-WebSocket-Key 字段的值，然后再对其结果通过 SHA1 哈希算法求出的结果。
    Sec-WebSocket-Protocol: chat  #对应于请求中的 Sec-WebSocket-Protocol 字段；

    2.3 WebSocket 优缺点

            优点：

            1、使用资源少。创建连接后，数据叫唤的包头较少；

            2、能实现及时通信。长连接，实时通信；

            3、更好的二进制支持。能更好的处理二进制内容；

            4、支持拓展。用户可以拓展协议，实现部分自定义的子协议。

            缺点：

            1、使用WebSocket，长连接，会占用一定资源；

            2、浏览器品类多，支持程度不同，可能问题多；

            3、与poll相比，代码复杂度将上升，完全依赖于websocket，要多写逻辑对websocket状态进行监控，对开发者要求也会高一些。

           没有完美的事物，我们讨论优缺点的目的是它适合什么场景，在要求实时性较高的应用时，那么WebSocket会更适合。如果基本都是操作的应用，实时性要求很低，那么WebSocket使用的资源成本就是不合适的。

    2.4 浏览器支持

          WebSocket - Web API 接口参考 | MDNWebSocket 对象提供了用于创建和管理 WebSocket 连接，以及可以通过该连接发送和接收数据的 API。icon-default.png?t=N7T8https://developer.mozilla.org/zh-CN/docs/Web/API/WebSocket以上是供参考的API接口，本文不做赘述，自己进入使用。

    三、前端使用示例

    文件名称

    具体代码

    1. /*
    2.  * @Descripttion: 封装socket方法
    3.  * @version:
    4.  * @Date: 2021-08-06 11:14:39
    5.  * @LastEditTime: 2021-10-26 14:06:34
    6.  */
    7. import { store } from "../store";
    8. import { ElMessage } from "element-plus";
    9. import { Base64 } from "js-base64";
    10. import { updateComputerIsValid } from "/@/service/AppService";
    11. import { localIp, playDutySound } from "../CommonUtil";
    12. import { deal3004Procotol, deal3005Procotol, deal3006Procotol } from "/@/service/WebsocketService"; //业务代码
    13.  
    14. interface socket {
    15.     websocket: any;
    16.     connectURL: string;
    17.     socket_open: boolean;
    18.     hearbeat_timer: any;
    19.     hearbeat_interval: number;
    20.     is_reonnect: boolean;
    21.     reconnect_count: number;
    22.     reconnect_current: number;
    23.     ronnect_number: number;
    24.     reconnect_timer: any;
    25.     reconnect_interval: number;
    26.     // init: (receiveMessage: Function | null) => any;
    27.     init: () => any;
    28.     receive: (message: any) => void;
    29.     heartbeat: () => void;
    30.     send: (data: any, callback?: any) => void;
    31.     close: () => void;
    32.     reconnect: () => void;
    33.     webSocketBack?: (message: any) => void;
    34. }
    35.  
    36. const socket: socket = {
    37.     websocket: null,
    38.     connectURL: import.meta.env.VITE_WEBSOCKET_MONITOR_URL + localIp().replaceAll(".", ""),
    39.     // 开启标识
    40.     socket_open: false,
    41.     // 心跳timer
    42.     hearbeat_timer: null,
    43.     // 心跳发送频率
    44.     hearbeat_interval: 3000,
    45.     // 是否自动重连
    46.     is_reonnect: true,
    47.     // 重连次数
    48.     reconnect_count: 5000,
    49.     // 已发起重连次数
    50.     reconnect_current: 1,
    51.     // 网络错误提示此时
    52.     ronnect_number: 0,
    53.     // 重连timer
    54.     reconnect_timer: null,
    55.     // 重连频率 不能设置的太小，否则会出现一次重连未返回的时候，下一次又开始重连
    56.     reconnect_interval: 6000,
    57.  
    58.     // init: (receiveMessage: Function | null) => {
    59.     init: () => {
    60.         if (!("WebSocket" in window)) {
    61.             // if (!("WebSocket" in window)) {
    62.             ElMessage.warning("浏览器不支持WebSocket");
    63.             return null;
    64.         }
    65.         // 已经创建过连接不再重复创建
    66.         if (socket.websocket) {
    67.             return socket.websocket;
    68.         }
    69.  
    70.         socket.websocket = new WebSocket(socket.connectURL);
    71.         socket.websocket.onmessage = (e: any) => {
    72.             // if (receiveMessage) {
    73.             //     receiveMessage(e);
    74.             // }
    75.             if (socket.webSocketBack) {
    76.                 socket.webSocketBack(e);
    77.             }
    78.         };
    79.  
    80.         socket.websocket.onclose = (e: any) => {
    81.             console.log("websocket--关闭", socket.reconnect_current,e);
    82.             if (socket.hearbeat_timer) {
    83.                 clearInterval(socket.hearbeat_timer);
    84.             }
    85.             //业务代码- 置位为1
    86.             updateComputerIsValid(localIp(), 1);
    87.             socket.socket_open = false;
    88.  
    89.             // 需要重新连接
    90.             if (socket.is_reonnect) {
    91.                 console.log("websocket--需要重新连接", socket.is_reonnect,socket.reconnect_interval);
    92.                 socket.reconnect_timer = setTimeout(() => {
    93.                     console.log("websocket--重连", socket.reconnect_current);
    94.                     // 超过重连次数
    95.                     if (
    96.                         socket.reconnect_current > socket.reconnect_count &&
    97.                         socket.reconnect_count > -1
    98.                     ) {
    99.                         console.log("websocket--超过重连次数");
    100.                         clearTimeout(socket.reconnect_timer);
    101.                         socket.is_reonnect = false;
    102.                         return;
    103.                     }
    104.  
    105.                     // 记录重连次数
    106.                     socket.reconnect_current++;
    107.                     //清除 socket.websocket
    108.                     socket.websocket = null;
    109.                     socket.reconnect();
    110.                 }, socket.reconnect_interval);
    111.             }
    112.         };
    113.  
    114.         // 连接成功
    115.         socket.websocket.onopen = function () {
    116.             console.log("websocket--连接成功");
    117.             //业务代码
    118.             updateComputerIsValid(localIp(), 0);
    119.             socket.socket_open = true;
    120.             socket.is_reonnect = true;
    121.             // 开启心跳
    122.             socket.heartbeat();
    123.         };
    124.  
    125.         // 连接发生错误
    126.         socket.websocket.onerror = function () {
    127.             console.log("websocket--发生错误!关闭执行重连");
    128.             socket.websocket.onclose();
    129.         };
    130.     },
    131.  
    132.     send: (data, callback = null) => {
    133.         // 开启状态直接发送
    134.         if (socket.websocket.readyState === socket.websocket.OPEN) {
    135.             socket.websocket.send(JSON.stringify(data));
    136.             if (callback) {
    137.                 callback();
    138.             }
    139.  
    140.             // 正在开启状态，则等待1s后重新调用
    141.         } else {
    142.             clearInterval(socket.hearbeat_timer);
    143.             if (socket.ronnect_number < 1) {
    144.                 // ElMessage({
    145.                 // 	type: 'error',
    146.                 // 	message: i18n.global.t('chat.unopen'),
    147.                 // 	duration: 0,
    148.                 // })
    149.                 console.log("服务关闭了！");
    150.             }
    151.             socket.ronnect_number++;
    152.         }
    153.     },
    154.  
    155.     receive: (message: any) => {
    156.         let params = Base64.decode(JSON.parse(message.data).data);
    157.         params = JSON.parse(params);
    158.         return params;
    159.     },
    160.  
    161.     heartbeat: () => {
    162.         if (socket.hearbeat_timer) {
    163.             clearInterval(socket.hearbeat_timer);
    164.         }
    165.  
    166.         socket.hearbeat_timer = setInterval(() => {
    167.             let diffMs = Number(new Date()) - Number(store.state.webSocketLastTime);
    168.             console.log("websocket--上次间隔时间:", diffMs, "3秒以上才发送心跳包");
    169.             if (diffMs > 0) {
    170.                 let data = {
    171.                     // languageId: store.state.users.language,
    172.                     content: "ping",
    173.                 };
    174.                 var sendDara = {
    175.                     encryption_type: "base64",
    176.                     data: Base64.encode(JSON.stringify(data)),
    177.                 };
    178.                 socket.send(sendDara);
    179.                 store.commit("setWebSocketLastTime", new Date());
    180.                 console.log(
    181.                     "websocket--心跳发送",
    182.                     sendDara,
    183.                     "更新时间：",
    184.                     store.state.webSocketLastTime
    185.                 );
    186.             }
    187.         }, socket.hearbeat_interval);
    188.     },
    189.  
    190.     close: () => {
    191.         clearInterval(socket.hearbeat_timer);
    192.         socket.is_reonnect = false;
    193.         socket.websocket.onclose();
    194.     },
    195.  
    196.     /**
    197.      * 重新连接
    198.      */
    199.     reconnect: () => {
    200.         //websocket存在且不想重连的时候
    201.         if (!socket.is_reonnect) {
    202.         // if (socket.websocket && !socket.is_reonnect) {
    203.             console.log("websocket--存在但是不需要重连的时候，关闭", socket.websocket, socket.is_reonnect);
    204.             socket.close();
    205.         }
    206.         
    207.         socket.init();
    208.     },
    209.  
    210.     /**
    211.      * 业务代码--数据处理 
    212.      * @param backMessage
    213.      */
    214.     webSocketBack(backMessage: any) {
    215.         store.commit("setWebSocketLastTime", new Date());
    216.         console.log(
    217.             "websocket-接受到的信息" + JSON.stringify(backMessage),
    218.             "更新的时间:",
    219.             store.state.webSocketLastTime
    220.         );
    221.         const wsData = backMessage.data.split("|");
    222.         const wsDataCode = backMessage.data.split("|")[0];
    223.         // 零位是协议号
    224.         switch (wsDataCode) {
    225.             // 值班机获取 提醒间隔时间 后的处理
    226.             case "3002": {
    227.                 console.log("收到ws:3002: " + JSON.stringify(wsData));
    228.                 const setHost = wsData[1];
    229.                 store.commit("setDutyConfirmTime", Number(wsData[2]));
    230.                 if (setHost === "0") {
    231.                     store.commit("setLocalComputerDutyState", 0);
    232.                     store.commit("setOnDutyState", 0);
    233.                 } else {
    234.                     store.commit("setLocalComputerDutyState", 1);
    235.                     store.commit("setOnDutyState", 1);
    236.                 }
    237.                 break;
    238.             }
    239.             case "3003": {
    240.                 console.log("收到ws:3003", wsDataCode);
    241.                 if (wsData[1] === "0") {
    242.                     playDutySound();
    243.                     if (store.state.onDutyState === 0) {
    244.                     } else if (store.state.onDutyState === 1) {、
    245.                         playDutySound();
    246.                         store.commit("setOnDutyState", true);
    247.                     }
    248.                 } else if (wsData[1] === "1") {
    249.                     store.commit("setOnDutyState", false);
    250.                 }
    251.                 break;
    252.             }
    253.             case "3004": {
    254.                 //更新store中的数据
    255.                 deal3004Procotol(wsData);
    256.                 break;
    257.             }
    258.             case "3005": {
    259.                 //更新store中的数据
    260.                 deal3005Procotol(wsData);
    261.                 break;
    262.             }
    263.             case "3006": {
    264.                 //更新store中的数据
    265.                 deal3006Procotol(wsData);
    266.                 break;
    267.             }
    268.         }
    269.     },
    270. };
    271.  
    272. export default socket;

    其中业务代码请不用关注，自己实现自己的业务逻辑即可。

    1001错误

    对于重连时间的设置，如果设置的时间太短，会出现反复1001错误（The WebSocket session [] timeout expired）关闭再重连的现象：

    把服务端关闭后

    每次错误返回中间有两次重连操作，所以调整了重连间隔时间，错误消失，推论：一次重连结果还未出来的时候，又发起了地址一样的连接请求，造成冲突，会关闭上次连接，这次关闭会引发上次连接的重连，这就造成了反复重连。目前我采用的是拉长重连时间，比较简单，可以尝试通过判断连接状态来阻止一次连接没完成之前再次连接。

    流程图

    启动连接

    1. //APP.VUE
    2. import socket from "/@/utils/websocket";
    3.  
    4. onMounted(async () => {
    5.   socket.init();
    6. });

    四、后端服务

    引入依赖包

    1. <dependency>
    2.     
    3.     <groupId>org.springframework.bootgroupId>
    4.     <artifactId>spring-boot-starter-websocketartifactId>
    5. dependency>

    客户端ip获取：

    在mainApplication上添加下面注解：

    @ServletComponentScan("**.**.filter")  //防止 @WebListener 无效
    1. import javax.servlet.ServletResponse;
    2. import javax.servlet.http.HttpServletRequest;
    3. import java.io.IOException;
    4.  
    5. @javax.servlet.annotation.WebFilter(filterName = "sessionFilter",urlPatterns = "/*")
    6. @Order(1)
    7. public class WebFilter implements Filter {
    8.     @Override
    9.     public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
    10.         HttpServletRequest req= (HttpServletRequest) servletRequest;
    11.         req.getSession().setAttribute("ip",req.getRemoteHost());
    12.         filterChain.doFilter(servletRequest,servletResponse);
    13.     }
    14. }

    WebSocket配置类

    在这里也做了IP的获取

    1. @Configuration
    2. public class WebSocketConfig extends ServerEndpointConfig.Configurator {
    3.     /**
    4.      * 注入一个ServerEndpointExporter,该Bean会自动注册使用@ServerEndpoint注解申明的websocket endpoint
    5.      */
    6.     @Bean
    7.     public ServerEndpointExporter serverEndpointExporter() {
    8.         return new ServerEndpointExporter();
    9.     }
    10.  
    11.  
    12.     @Override
    13.     public void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request, HandshakeResponse response) {
    14.  
    15.         Map attributes = sec.getUserProperties();
    16.         HttpSession session = (HttpSession) request.getHttpSession();
    17.         if (session != null) {
    18.             attributes.put(GlobalContants.IP_ADDR, session.getAttribute("ip"));
    19.             Enumeration names = session.getAttributeNames();
    20.             while (names.hasMoreElements()) {
    21.                 String name = names.nextElement();
    22.                 attributes.put(name, session.getAttribute(name));
    23.             }
    24.         }
    25.     }
    26. }

    Websocket接收

    1. import com.baomidou.mybatisplus.core.toolkit.ArrayUtils;
    2. import com.deyou.cabin.monitor.common.GlobalContants;
    3. import com.deyou.cabin.monitor.common.GlobalParams;
    4. import com.deyou.cabin.monitor.common.utils.AssembleDownProtocolUtils;
    5. import com.deyou.cabin.monitor.common.utils.CommonServeUtils;
    6. import com.deyou.cabin.monitor.config.WebSocketConfig;
    7. import com.deyou.cabin.monitor.model.WebSocketModel;
    8. import lombok.extern.slf4j.Slf4j;
    9. import org.apache.commons.lang3.StringUtils;
    10. import org.springframework.stereotype.Component;
    11.  
    12. import javax.websocket.OnClose;
    13. import javax.websocket.OnError;
    14. import javax.websocket.OnMessage;
    15. import javax.websocket.OnOpen;
    16. import javax.websocket.Session;
    17. import javax.websocket.server.PathParam;
    18. import javax.websocket.server.ServerEndpoint;
    19. import java.time.LocalDateTime;
    20. import java.util.ArrayList;
    21. import java.util.List;
    22. import java.util.Map;
    23.  
    24.  
    25. @Slf4j
    26. @Component
    27. //@RequiredArgsConstructor
    28. @ServerEndpoint(value = "/websocket/monitor/{code}",configurator = WebSocketConfig.class)
    29. public class WebsocketController {
    30.  
    31.     /**
    32.      * 连接建立成功调用的方法
    33.      */
    34.     @OnOpen
    35.     public void onOpen(Session session,@PathParam(value = "code") String code) {
    36.         try
    37.         {
    38.             session.setMaxIdleTimeout(30000);
    39.         }
    40.         catch (Exception e)
    41.         {
    42.             log.error(e.getMessage(),e);
    43.         }
    44.     }
    45.  
    46.     /**
    47.      * 连接关闭调用的方法
    48.      */
    49.     @OnClose
    50.     public void onClose(Session session) {
    51.         try
    52.         {
    53.             log.info("有一连接关闭：{}，当前在线人数为：{}", session.getId(), 
    54.         }
    55.         catch (Exception e)
    56.         {
    57.             log.error(e.getMessage(),e);
    58.         }
    59.     }
    60.  
    61.     /**
    62.      * 收到客户端消息后调用的方法
    63.      *
    64.      * @param message
    65.      *            
    66.      */
    67.     @OnMessage
    68.     public void onMessage(String message, Session session) {
    69.         try
    70.         {
    71.             log.info("服务端收到客户端[{}]的消息:{}", session.getId(), message);
    72.         }
    73.         catch (Exception ex)
    74.         {
    75.             log.error(ex.getMessage(),ex);
    76.         }
    77.         
    78.     }
    79.  
    80.     @OnError
    81.     public void onError(Session session, Throwable error) {
    82.         log.error("websocket发生错误:" + session.getId() + "---" + error.getMessage(),error);
    83.     }
    84.  
    85.     public void sendMessageToAll(String message, Session fromSession) {
    86.         try
    87.         {
    88.             //GlobalParams.webSocketModelMap是全局变量 ConcurrentHashMap webSocketModelMap
    89.             for (Map.Entry sessionEntry : GlobalParams.webSocketModelMap.entrySet()) {
    90.                 Session toSession = sessionEntry.getValue().getSession();
    91.                 // 排除掉自己
    92.                 if (!fromSession.getId().equals(toSession.getId())) {
    93.                     log.info("服务端给客户端[{}][{}]发送消息{}", toSession.getId(),
    94.                             sessionEntry.getValue().getWebSocketCode(), message);
    95.                     sendMessToOne(message,toSession);
    96.                 }
    97.             }
    98.         }
    99.         catch (Exception e)
    100.         {
    101.             log.error(e.getMessage(),e);
    102.         }
    103.  
    104.     }
    105.  
    106.     public void sendMessageToAll(String message) {
    107.         try {
    108.             //GlobalParams.webSocketModelMap是全局变量 ConcurrentHashMap webSocketModelMap
    109.             for (Map.Entry sessionEntry : GlobalParams.webSocketModelMap.entrySet()) {
    110.                 Session toSession = sessionEntry.getValue().getSession();
    111.                 log.info("服务端给客户端[{}][{}]发送消息{}", toSession.getId(), sessionEntry.getValue().getWebSocketCode(), message);
    112.                 sendMessToOne(message, toSession);
    113.             }
    114.         } catch (Exception e) {
    115.             log.error(e.getMessage(), e);
    116.         }
    117.     }
    118.  
    119.     public void sendMessToOne(String message, Session toSession) {
    120.         try {
    121.             // 尝试过锁住方法，还是不行，这里锁住webSocketMap，让多线程，访问不同对象，也能同步
    122.             synchronized(GlobalParams.webSocketModelMap){
    123.                 String toId = toSession.getId();
    124.                 if (StringUtils.isNotBlank(toId) && GlobalParams.webSocketModelMap.containsKey(toId)) {
    125.                     GlobalParams.webSocketModelMap.get(toId).getSession().getBasicRemote().sendText(message);
    126.                 }
    127.             }
    128.         } catch (Exception e) {
    129.             log.error("服务端发送消息给客户端失败,"+e.getMessage(),e);
    130.         }
    131.     }
    132.  
    133. }

    其中，synchronized(GlobalParams.webSocketModelMap)中GlobalParams.webSocketModelMap是我记录当前在线的websocket的信息。上边代码的注释中已经写了，这个锁的目的是为了解决websocket服务端下发时出现的错误“The remote endpoint was in state [STREAM_WRITING] which is an invalid state for called method”的错误，问题的引发场景和分析个人记录如下： 

    1.因为在 @OnMessage中，我有两个方法同时使用了session，导致session多线程不安全，发生的频次少都可能不出现这个问题！

    2.JSON.toJSONString(GlobalParams.webSocketModelMap) 其中带有session，会引发这个问题 解决办法：加异步锁，但是需要锁定 ConcurrentHashMap。

    使用

    1. @Resource
    2. private WebsocketController websocketService;
    1. try{
    2.     websocketService.sendMessToOne(sendMes, toSession);
    3. }catch (Exception e){
    4.     log.error(e.getMessage(),e);
    5. }

    5、结束

    连接地址：ws://IP:PORT/websocket/monitor/{code} ，其中code是你自己定义的值。

  • 相关阅读:
    Private market：借助ZK实现的任意计算的trustless交易
    22. containerd使用Devmapper snapshotter讲解
    卡尔曼及扩展卡尔曼滤波详细推导-来自DR_CAN视频
    antv/x6 键盘快捷键事件
    残差resnet复现，源代码理解
    为什么用Selenium做自动化测试，你真的知道吗？
    有趣的前端面试题
    CURL简单使用
    docker启动镜像命令
    unity 点击3D物体
  • 原文地址：https://blog.csdn.net/hchyboy/article/details/132927047