【Java】SpringBoot使用websocket完成与页面实时通讯（下发通知）

浅浅的花香味﹌ 2023-10-15 10:56 15阅读 0赞

### 前言： ###

之前接手的一个后台管理系统项目中，有下发通知功能，一直使用的是ajax循环请求获取最新通知列表，导致无效请求过多。决定优化下，使用websocket连接来实时通知页面更新通知列表。以下是实现方式及过程中遇到的问题。

### 什么是socket： ###

websocket是HTML5开始提供的一种客户端与服务器之间进行通讯的网络技术,通过这种方式可以实现客户端和服务器的长连接,双向实时通讯。你可以将它看做是实现网络通信的接口，让应用程序能够互相发送和接收数据。Socket 有两种主要类型：**TCP（传输控制协议）和 UDP（用户数据报协议）**，解决的问题和场景略有区别。

优点：减少资源消耗;实时推送不用等待客户端的请求;减少通信量;  
  缺点：少部分浏览器不支持，不同浏览器支持的程度和方式都不同。  
  应用场景：聊天室、智慧大屏、消息提醒、股票k线图监控等。

下面简述一下 Socket 的工作原理：  
![在这里插入图片描述][14cc1014e33947448c14c528d1498636.png]

1.  套接字通信的基本单位是两个套接字，分别位于通信程序的一端。客户端套接字与服务器端套接字通过网络连接进行通信。
2.  在通信开始之前，服务器端需要创建一个套接字，绑定一个地址和端口，并开始监听客户端的连接请求。
3.  客户端也需要创建一个套接字，并尝试连接到服务器的地址和端口。若成功建立连接，通信开始。
4.  之后，客户端和服务器都可以通过发送或接收数据来进行通信。这里的数据发送和接收是实时的，并且通常会采用字节流的形式。
5.  通信完成后，客户端和服务器需要关闭套接字，释放资源。

上面的描述是基于 TCP 协议的 Socket 通信原理。TCP 是一种面向连接的协议，特点是能够保证数据的可靠传输和顺序，并在网络拥塞时进行流量控制。而基于 UDP 的 Socket 通信则是无连接的，发送方直接将数据包发往接收方，具有更低的延迟，但可能会出现丢包和乱序的情况。可以根据具体场景和需求选择合适的协议类型。  
本文使用的是TCP协议

### 代码实现 ###

#### 一：后台代码 ####

**1. 引入依赖**

<dependency>
    	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>

**2. 添加config配置**

package org.jeecg.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    /**
     * @Description: WebSocketConfig
     * @author: jeecg-boot
     */
    @Configuration
    public class WebSocketConfig {
        
        /**
         * 	注入ServerEndpointExporter，
         * 	这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
         */
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
        
            return new ServerEndpointExporter();
        }
    
    }

**3. 编写websocket实现类**

package org.jeecg.modules.message.websocket;
    
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import javax.annotation.Resource;
    import javax.websocket.*;
    import javax.websocket.server.PathParam;
    import javax.websocket.server.ServerEndpoint;
    
    import com.alibaba.fastjson.JSONObject;
    import org.jeecg.common.base.BaseMap;package org.jeecg.modules.message.websocket;
    
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import javax.websocket.*;
    import javax.websocket.server.PathParam;
    import javax.websocket.server.ServerEndpoint;
    
    import com.alibaba.fastjson.JSONObject;
    import org.jeecg.common.constant.WebsocketConst;
    import org.jeecg.common.modules.redis.client.JeecgRedisClient;
    import org.springframework.stereotype.Component;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @Author scott
     * @Date 2019/11/29 9:41
     * @Description: 此注解相当于设置访问URL
     */
    @Component
    @Slf4j
    @ServerEndpoint("/websocket/{userId}")
    public class WebSocket {
        
        
        /**线程安全Map*/
        private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    
        //==========【websocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
        @OnOpen
        public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        
            try {
        
                sessionPool.put(userId, session);
                log.info("【系统 WebSocket】有新的连接，总数为:" + sessionPool.size());
            } catch (Exception e) {
        
            }
        }
    
        @OnClose
        public void onClose(@PathParam("userId") String userId) {
        
            try {
        
                sessionPool.remove(userId);
                log.info("【系统 WebSocket】连接断开，总数为:" + sessionPool.size());
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    
        /**
         * ws推送消息
         *
         * @param userId
         * @param message
         */
        public void pushMessage(String userId, String message) {
        
            for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
        
                //userId key值= {用户id + "_"+ 登录token的md5串}
                //TODO vue2未改key新规则，暂时不影响逻辑
                if (item.getKey().contains(userId)) {
        
                    Session session = item.getValue();
                    try {
        
                        //update-begin-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
                        synchronized (session){
        
                            log.info("【系统 WebSocket】推送单人消息:" + message);
                            session.getBasicRemote().sendText(message);
                        }
                        //update-end-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
                    } catch (Exception e) {
        
                        log.error(e.getMessage(),e);
                    }
                }
            }
        }
    
        /**
         * ws遍历群发消息
         */
        public void pushMessage(String message) {
        
            try {
        
                for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
        
                    try {
        
                        item.getValue().getAsyncRemote().sendText(message);
                    } catch (Exception e) {
        
                        log.error(e.getMessage(), e);
                    }
                }
                log.info("【系统 WebSocket】群发消息:" + message);
            } catch (Exception e) {
        
                log.error(e.getMessage(), e);
            }
        }
    
    
        /**
         * ws接受客户端消息
         */
        @OnMessage
        public void onMessage(String message, @PathParam(value = "userId") String userId) {
        
            if(!"ping".equals(message) && !WebsocketConst.CMD_CHECK.equals(message)){
        
                log.info("【系统 WebSocket】收到客户端消息:" + message);
            }else{
        
                log.debug("【系统 WebSocket】收到客户端消息:" + message);
            }
            
            //------------------------------------------------------------------------------
            JSONObject obj = new JSONObject();
            //业务类型
            obj.put(WebsocketConst.MSG_CMD, WebsocketConst.CMD_CHECK);
            //消息内容
            obj.put(WebsocketConst.MSG_TXT, "心跳响应");
            this.pushMessage(userId, obj.toJSONString());
            //------------------------------------------------------------------------------
        }
    
        /**
         * 配置错误信息处理
         *
         * @param session
         * @param t
         */
        @OnError
        public void onError(Session session, Throwable t) {
        
            log.warn("【系统 WebSocket】消息出现错误");
            //t.printStackTrace();
        }
        //==========【系统 WebSocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
        
    import org.jeecg.common.constant.WebsocketConst;
    import org.springframework.stereotype.Component;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @Author scott
     * @Date 2019/11/29 9:41
     * @Description: 此注解相当于设置访问URL
     */
    @Component
    @Slf4j
    @ServerEndpoint("/websocket/{userId}")
    public class WebSocket {
        
        
        /**线程安全Map*/
        private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();
    
        /**
         * Redis触发监听名字
         */
        public static final String REDIS_TOPIC_NAME = "socketHandler";
        @Resource
        private JeecgRedisClient jeecgRedisClient;
    
    
        //==========【websocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
        @OnOpen
        public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        
            try {
        
                sessionPool.put(userId, session);
                log.info("【系统 WebSocket】有新的连接，总数为:" + sessionPool.size());
            } catch (Exception e) {
        
            }
        }
    
        @OnClose
        public void onClose(@PathParam("userId") String userId) {
        
            try {
        
                sessionPool.remove(userId);
                log.info("【系统 WebSocket】连接断开，总数为:" + sessionPool.size());
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    
        /**
         * ws推送消息
         *
         * @param userId
         * @param message
         */
        public void pushMessage(String userId, String message) {
        
            for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
        
                //userId key值= {用户id + "_"+ 登录token的md5串}
                if (item.getKey().contains(userId)) {
        
                    Session session = item.getValue();
                    try {
        
                        //update-begin-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
                        synchronized (session){
        
                            log.info("【系统 WebSocket】推送单人消息:" + message);
                            session.getBasicRemote().sendText(message);
                        }
                        //update-end-author:taoyan date:20211012 for: websocket报错 https://gitee.com/jeecg/jeecg-boot/issues/I4C0MU
                    } catch (Exception e) {
        
                        log.error(e.getMessage(),e);
                    }
                }
            }
        }
    
        /**
         * ws遍历群发消息
         */
        public void pushMessage(String message) {
        
            try {
        
                for (Map.Entry<String, Session> item : sessionPool.entrySet()) {
        
                    try {
        
                        item.getValue().getAsyncRemote().sendText(message);
                    } catch (Exception e) {
        
                        log.error(e.getMessage(), e);
                    }
                }
                log.info("【系统 WebSocket】群发消息:" + message);
            } catch (Exception e) {
        
                log.error(e.getMessage(), e);
            }
        }
    
    
        /**
         * ws接受客户端消息
         */
        @OnMessage
        public void onMessage(String message, @PathParam(value = "userId") String userId) {
        
            if(!"ping".equals(message) && !WebsocketConst.CMD_CHECK.equals(message)){
        
                log.info("【系统 WebSocket】收到客户端消息:" + message);
            }else{
        
                log.debug("【系统 WebSocket】收到客户端消息:" + message);
            }
            
            //------------------------------------------------------------------------------
            JSONObject obj = new JSONObject();
            //业务类型
            obj.put(WebsocketConst.MSG_CMD, WebsocketConst.CMD_CHECK);
            //消息内容
            obj.put(WebsocketConst.MSG_TXT, "心跳响应");
            this.pushMessage(userId, obj.toJSONString());
            //------------------------------------------------------------------------------
        }
    
        /**
         * 配置错误信息处理
         *
         * @param session
         * @param t
         */
        @OnError
        public void onError(Session session, Throwable t) {
        
            log.warn("【系统 WebSocket】消息出现错误");
            //t.printStackTrace();
        }
        //==========【系统 WebSocket接受、推送消息等方法 —— 具体服务节点推送ws消息】========================================================================================
    }

为方便理解，加入了gpt对这段代码的解析。

> 这段代码是一个 Java 类，用于创建一个 WebSocket 服务器端点。WebSocket是一种用于在浏览器和服务器之间进行双向实时通信的技术。以下是代码的主要部分和功能的概述：
> 
> 1.  引入依赖：引入了一些必要的 Java 类和类库，例如 WebSocket 相关的类、ConcurrentHashMap、JSONObject等。
> 2.  类声明与注解：WebSocket 类被注解为一个组件（@Component）以及一个 WebSocket服务器端点（@ServerEndpoint(“/websocket/\{userId\}”)）。服务器端点 URL 为"/websocket/\{userId\}"，其中 “\{userId\}” 是 URL 路径参数。
> 3.  变量声明：定义了一个线程安全的 ConcurrentHashMap 变量 sessionPool，用于储存与每个用户相关的WebSocket 会话对象。
> 4.  连接打开（onOpen 方法）：当一个新的 WebSocket 连接被建立时，这个方法会被调用。方法将用户ID作为键，将新的WebSocket 会话添加到 sessionPool 中。
> 5.  连接关闭（onClose 方法）：当一个 WebSocket 连接断开时，这个方法会被调用。方法将用户ID对应的 WebSocket 会话从sessionPool 中移除。
> 6.  推送消息（pushMessage 方法）：用于向指定用户推送消息。方法遍历sessionPool，找到与指定用户ID匹配的会话，并将消息发送给这个用户。
> 7.  群发消息（pushMessage 方法）：用于向所有已连接的用户推送消息。方法遍历 sessionPool，并向每个会话发送消息。
> 8.  接收客户端消息（onMessage方法）：当服务器接收到客户端发送的消息时，这个方法会被调用。方法首先对收到的消息进行日志记录，然后构造一个心跳响应消息，并将这个消息推送给发送该消息的用户。
> 9.  错误处理（onError 方法）：当 WebSocket 会话出现错误时，这个方法会被调用。方法仅记录错误日志，不进行其他处理。
> 
> 这段代码实现了一个实时通信的 WebSocket服务器端点，并提供了基本的消息推送功能，包括一对一消息推送和群发消息。通过这个端点，客户端可以与服务器进行实时通信，发送和接收消息

至此后台代码完成。

#### 二：页面代码 ####

#### 三：过程中问题 ####

**1. ws请求401**  
原因：项目中使用了Shiro权限管理，将WebSocket请求添加拦截了。将socket请求url添加到不拦截列表即可。  
**这里的示例是Shiro。** 在Config配置文件中，将websocket请求排除拦截。配置如下

/**
         * Filter Chain定义说明
         *
         * 1、一个URL可以配置多个Filter，使用逗号分隔
         * 2、当设置多个过滤器时，全部验证通过，才视为通过
         * 3、部分过滤器可指定参数，如perms，roles
         */
        @Bean("shiroFilterFactoryBean")
        public ShiroFilterFactoryBean shiroFilter(SecurityManager securityManager) {
        
            CustomShiroFilterFactoryBean shiroFilterFactoryBean = new CustomShiroFilterFactoryBean();
            shiroFilterFactoryBean.setSecurityManager(securityManager);
            // 拦截器
            Map<String, String> filterChainDefinitionMap = new LinkedHashMap<String, String>();
    
            //支持yml方式，配置拦截排除
            if(jeecgBaseConfig!=null && jeecgBaseConfig.getShiro()!=null){
        
                String shiroExcludeUrls = jeecgBaseConfig.getShiro().getExcludeUrls();
                if(oConvertUtils.isNotEmpty(shiroExcludeUrls)){
        
                    String[] permissionUrl = shiroExcludeUrls.split(",");
                    for(String url : permissionUrl){
        
                        filterChainDefinitionMap.put(url,"anon");
                    }
                }
            }
            filterChainDefinitionMap.put("/websocket/**", "anon");//系统通知和公告
            // 未授权界面返回JSON
            shiroFilterFactoryBean.setUnauthorizedUrl("/sys/common/403");
            shiroFilterFactoryBean.setLoginUrl("/sys/common/403");
            shiroFilterFactoryBean.setFilterChainDefinitionMap(filterChainDefinitionMap);
            return shiroFilterFactoryBean;
       }

**2. 本地测试ok，上线socket初始化失败**  
需在项目server下添加此nginx配置，proxy\_pass 配置为自己后台项目路径

#配置wss开始-----/jeecg-boot/websocket/ 路径后还有参数配置如下：----------
           location ^~ /jeecg-boot/websocket/  {
                            proxy_pass http://192.168.2.77:8080/jeecg-boot/websocket/;
    
                            proxy_set_header X-Real-IP $remote_addr;
                            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
                            proxy_set_header Host $host;
                            proxy_set_header X-NginX-Proxy true;
                            proxy_http_version 1.1;
                            proxy_set_header Upgrade $http_upgrade;
                            proxy_set_header Connection "Upgrade";
                            proxy_connect_timeout 600s;
                            proxy_read_timeout 600;
                            proxy_send_timeout 600s;
           }
           #配置wss结束------------------------------------------

需在项目server平级位置添加此nginx配置

map $http_upgrade $connection_upgrade {
        default upgrade;
        '' close;
    }

#### 四：效果展示 ####

![在这里插入图片描述][9b0f052c25b940bcb14e527adb676726.png]  
此时，页面可以实时更新通知消息列表。

本篇文章至此结束，有问题欢迎留言区留言或私信，看到后会第一时间帮忙解决。多谢观看。

[14cc1014e33947448c14c528d1498636.png]: https://img-blog.csdnimg.cn/14cc1014e33947448c14c528d1498636.png
[9b0f052c25b940bcb14e527adb676726.png]: https://img-blog.csdnimg.cn/9b0f052c25b940bcb14e527adb676726.png