Springboot集成websocket

Springboot集成websocket

码农世界 2024-06-21 后端 97 次浏览 0个评论

文章目录

  • Springboot集成websocket
    • 1. WebSocket概述
    • 2. WebSocket原理
    • 3. Spring Boot集成WebSocket
      • 3.1 添加依赖
      • 3.2 创建WebSocket配置类
      • 3.3 创建消息处理器
      • 3.4 服务器主动给客户端发送消息
      • 4. 使用ApiPost测试WebSocket
      • 5. 参考和感谢

        Springboot集成websocket

        1. WebSocket概述

        我们再日常的web应用开发过程中,常见的是前端向后端发起请求,但有的时候需要后端主动给前端发送消息,这个时候全双工的websocket即时通讯就闪亮登场了。

        Springboot集成websocket

        • 全双工通讯模式允许双方同时进行双向通讯,如手机通话。
        • 半双工通讯模式允许双方交替发送和接收信息,但不能同时通讯,如对讲机。
        • 单工通讯模式只能单向传输信息,不能回复,如广播电台。

          2. WebSocket原理

          WebSocket是一种在单个TCP连接上进行全双工通信的协议。它通过一个简单的握手过程来建立连接,然后在连接上进行双向数据传输。与传统的HTTP请求不同,WebSocket连接一旦建立,就可以在客户端和服务器之间保持打开状态,直到被任何一方关闭。

          Springboot集成websocket

          WebSocket协议的核心特点包括:

          • 全双工通信:客户端和服务器可以同时发送和接收消息。
          • 持久连接:一旦建立连接,就可以持续进行数据交换,无需像HTTP那样频繁地建立新的连接。
          • 低延迟:由于连接是持久的,数据可以几乎实时地发送和接收。
          • 轻量级协议:WebSocket协议的头部信息非常简单,减少了数据传输的开销。

            3. Spring Boot集成WebSocket

            在Spring Boot中集成WebSocket非常简单,Spring提供了对WebSocket的原生支持。以下是一个基本的集成步骤:

            3.1 添加依赖

            在pom.xml中添加Spring Boot的WebSocket依赖:

            
                org.springframework.boot
                spring-boot-starter-websocket
            
            

            3.2 创建WebSocket配置类

            创建一个配置类来启用和配置WebSocket:

            package com.jiayuan.common.config;
            import org.springframework.context.annotation.Bean;
            import org.springframework.context.annotation.Configuration;
            import org.springframework.web.socket.server.standard.ServerEndpointExporter;
            import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;
            /**
             * @Title: WebSocketConfig
             * @Package: com.jiayuan.common.config
             * @Description: websocket配置
             * @Author: xmc
             * @Date: 创建时间 2024-04-24
             */
            @Configuration
            public class WebSocketConfig {
                /**
                 * 自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
                 *
                 * @return
                 */
                @Bean
                public ServerEndpointExporter serverEndpointExporter() {
                    return new ServerEndpointExporter();
                }
                /**
                 * 通信文本消息和二进制缓存区大小
                 * 避免对接 第三方 报文过大时,Websocket 1009 错误
                 *
                 * @return
                 */
                @Bean
                public ServletServerContainerFactoryBean createWebSocketContainer() {
                    ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
                    // 在此处设置bufferSize
                    container.setMaxTextMessageBufferSize(10240000);
                    container.setMaxBinaryMessageBufferSize(10240000);
                    container.setMaxSessionIdleTimeout(15 * 60000L);
                    return container;
                }
            }
             
            

            3.3 创建消息处理器

            package com.jiayuan.common.config;
            import cn.hutool.core.collection.CollectionUtil;
            import cn.hutool.extra.spring.SpringUtil;
            import cn.hutool.json.JSONUtil;
            import com.alibaba.fastjson2.JSON;
            import com.alibaba.fastjson2.JSONObject;
            import com.baomidou.mybatisplus.core.toolkit.StringUtils;
            import com.jiayuan.common.redis.RedisCache;
            import com.jiayuan.modules.critical.dto.CvRecordExtraDTO;
            import com.jiayuan.modules.critical.dto.SyncRecordDTO;
            import com.jiayuan.modules.critical.service.CvRecordService;
            import lombok.RequiredArgsConstructor;
            import lombok.extern.slf4j.Slf4j;
            import org.springframework.beans.factory.annotation.Autowired;
            import org.springframework.stereotype.Component;
            import javax.annotation.Resource;
            import javax.websocket.*;
            import javax.websocket.server.PathParam;
            import javax.websocket.server.ServerEndpoint;
            import java.io.IOException;
            import java.util.List;
            import java.util.concurrent.ConcurrentHashMap;
            /**
             * @Title: WebSocketServer
             * @Package: com.jiayuan.common.config
             * @Description: websocket的服务端
             * @Author: xmc
             * @Date: 创建时间 2024-04-24
             */
            @Component
            @Slf4j
            @ServerEndpoint("/api/pushMessage/{userId}")
            public class WebSocketServer {
                /**
                 * 静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。
                 */
                private static int onlineCount = 0;
                /**
                 * concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。
                 */
                private static ConcurrentHashMap webSocketMap = new ConcurrentHashMap<>();
                /**
                 * 与某个客户端的连接会话,需要通过它来给客户端发送数据
                 */
                private Session session;
                /**
                 * 接收userId
                 */
                private String userId = "";
                /**
                 * 连接建立成
                 * 功调用的方法
                 */
                @OnOpen
                public void onOpen(Session session, @PathParam("userId") String userId) {
                    this.session = session;
                    this.userId = userId;
                    if (webSocketMap.containsKey(userId)) {
                        webSocketMap.remove(userId);
                        //加入set中
                        webSocketMap.put(userId, this);
                    } else {
                        //加入set中
                        webSocketMap.put(userId, this);
                        //在线数加1
                        addOnlineCount();
                    }
                    log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());
                    sendMessage("连接成功");
                }
                /**
                 * 连接关闭
                 * 调用的方法
                 */
                @OnClose
                public void onClose() {
                    if (webSocketMap.containsKey(userId)) {
                        webSocketMap.remove(userId);
                        //从set中删除
                        subOnlineCount();
                    }
                    log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());
                }
                /**
                 * 收到客户端消
                 * 息后调用的方法
                 *
                 * @param message 客户端发送过来的消息
                 **/
                @OnMessage
                public void onMessage(String message, Session session) {
                    log.info("用户消息:" + userId + ",报文:" + message);
                    //可以群发消息
                    //消息保存到数据库、redis
                    if (StringUtils.isNotBlank(message)) {
                        try {
                            //解析发送的报文
                            JSONObject jsonObject = JSON.parseObject(message);
                            //追加发送人(防止串改)
                            jsonObject.put("fromUserId", this.userId);
                            String toUserId = jsonObject.getString("toUserId");
                            //传送给对应toUserId用户的websocket
                            if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {
                                webSocketMap.get(toUserId).sendMessage(message);
                            } else {
                                //否则不在这个服务器上,发送到mysql或者redis
                                log.error("请求的userId:" + toUserId + "不在该服务器上");
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                /**
                 * @param session
                 * @param error
                 */
                @OnError
                public void onError(Session session, Throwable error) {
                    log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
                    error.printStackTrace();
                }
                /**
                 * 实现服务
                 * 器主动推送
                 */
                public void sendMessage(String message) {
                    try {
                        this.session.getBasicRemote().sendText(message);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                /**
                 * 发送自定
                 * 义消息
                 **/
                public static void sendInfo(String message, String userId) {
                    log.info("发送消息到:" + userId + ",报文:" + message);
                    if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
                        webSocketMap.get(userId).sendMessage(message);
                    } else {
                        log.error("用户" + userId + ",不在线!");
                    }
                }
                /**
                 * 获得此时的
                 * 在线人数
                 *
                 * @return
                 */
                public static synchronized int getOnlineCount() {
                    return onlineCount;
                }
                /**
                 * 在线人
                 * 数加1
                 */
                public static synchronized void addOnlineCount() {
                    WebSocketServer.onlineCount++;
                }
                /**
                 * 在线人
                 * 数减1
                 */
                public static synchronized void subOnlineCount() {
                    WebSocketServer.onlineCount--;
                }
            }
             
            

            上述代码可能会有疑问:session和userId两个字段是不是安全的?多人连接,后面的会不会覆盖掉前面session和userId。

            Springboot集成websocket

            答案是不会的,关于处理器WebSocketServer我们要明确以下几点:

            1. websocket是原型模式,@ServerEndpoint每次建立双向通信的时候都会创建一个实例
            2. 为什么每次都@OnOpen都要检查webSocketMap.containsKey(userId) ,实际使用的时候发现偶尔会出现重连失败或者其他原因导致之前的session还存在,这时候就需要一个清除动作

            3.4 服务器主动给客户端发送消息

             WebSocketServer.sendInfo("服务器主动给客户端发送消息test", "zhangsan");
            

            4. 使用ApiPost测试WebSocket

            以下是如何使用ApiPost进行测试的步骤:

            1. 新建一个websocket测试

            Springboot集成websocket

            1. 填写URLws://localhost:8080/cvms-api/api/pushMessage/3

            Springboot集成websocket

            Springboot集成websocket

            注意以下几点:

            • 协议是ws,加密方式请选择wss
            • 选择Raw
            • URL的拼接公式如下
              # servlet.context-path 这个是application.yml中的配置
              ws://ip:port//${servlet.context-path}/注解@ServerEndpoint的值
              
              • 有权限验证的,比如说shiro权限验证,URL就需要加入白名单
                 filterMap.put("/api/pushMessage/*", "anon");
                

                5. 参考和感谢

                SpringBoot2.0集成WebSocket,实现后台向前端推送信息

转载请注明来自码农世界,本文标题:《Springboot集成websocket》

百度分享代码,如果开启HTTPS请参考李洋个人博客
每一天,每一秒,你所做的决定都会改变你的人生!

发表评论

快捷回复:

评论列表 (暂无评论,97人围观)参与讨论

还没有评论,来说两句吧...

Top