SpringBoot整合WebSocket（session共享实现）

港控/mmm° 2023-09-30 10:49 14阅读 0赞

## 前言 ##

WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信——允许服务器主动发送信息给客户端。

##  ##

--------------------

一、为什么需要WebSocket

二、SpringBoot整合WebSocket

1.maven依赖

2.WebSocketConfig

3.WebSocketServer

4.集群版

4.1Redis发布订阅模式实现

4.2队列消息推送

4.3队列消息接收

--------------------

## **一、为什么需要****WebSocket** ##

HTTP 是基于请求响应式的，即通信只能由客户端发起，服务端做出响应，无状态，无连接。

无状态：每次连接只处理一个请求，请求结束后断开连接。

无连接：对于事务处理没有记忆能力，服务器不知道客户端是什么状态。

通过HTTP实现即时通讯，只能是页面轮询向服务器发出请求，服务器返回查询结果。轮询的效率低，非常浪费资源，因为必须不停连接，或者 HTTP 连接始终打开。

WebSocket的最大特点就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话。

WebSocket特点

（1）建立在 TCP 协议之上，服务器端的实现比较容易。

（2）与 HTTP 协议有着良好的兼容性。默认端口也是80和443，并且握手阶段采用 HTTP 协议，因此握手时不容易屏蔽，能通过各种 HTTP 代理服务器。

（3）数据格式比较轻量，性能开销小，通信高效。

（4）可以发送文本，也可以发送二进制数据。

（5）没有同源限制，客户端可以与任意服务器通信。

（6）协议标识符是ws（如果加密，则为wss），服务器网址就是 URL。

## 二、SpringBoot整合WebSocket ##

### 1.maven依赖 ###

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

### 2.WebSocketConfig ###

启用WebSocket的支持

@Configuration
    public class WebSocketConfig {
        /**
         * 自动注册使用@ServerEndpoint注解声明的websocket endpoint
         * 2022/2/14
         * @return
         */
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
    }

### 3.WebSocketServer ###

因为WebSocket是类似客户端服务端的形式(采用ws协议)，那么这里的WebSocketServer其实就相当于一个ws协议的Controller

直接@ServerEndpoint("/websocket/\{userId\}") 、@Component启用即可，然后在里面实现@OnOpen开启连接，@onClose关闭连接，@onMessage接收消息等方法。

集群版（多个ws节点）还需要借助mysql或者redis等进行处理，改造对应的sendMessage方法即可。

import java.io.IOException;
    import java.util.concurrent.ConcurrentHashMap;
    import javax.websocket.OnClose;
    import javax.websocket.OnError;
    import javax.websocket.OnMessage;
    import javax.websocket.OnOpen;
    import javax.websocket.Session;
    import javax.websocket.server.PathParam;
    import javax.websocket.server.ServerEndpoint;
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import org.apache.commons.lang.StringUtils;
    import org.springframework.stereotype.Component;
    import cn.hutool.log.Log;
    import cn.hutool.log.LogFactory;
    
    
    @ServerEndpoint("/imserver/{userId}")
    @Component
    public class WebSocketServer {
    
        static Log log=LogFactory.get(WebSocketServer.class);
        /**静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。*/
        private static int onlineCount = 0;
        /**concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。*/
        private static ConcurrentHashMap<String,WebSocketServer> 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);
                webSocketMap.put(userId,this);
                //加入set中
            }else{
                webSocketMap.put(userId,this);
                //加入set中
                addOnlineCount();
                //在线数加1
            }
    
            log.info("用户连接:"+userId+",当前在线人数为:" + getOnlineCount());
    
            try {
                sendMessage("连接成功");
            } catch (IOException e) {
                log.error("用户:"+userId+",网络异常!!!!!!");
            }
        }
    
        /**
         * 连接关闭调用的方法
         */
        @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(jsonObject.toJSONString());
                    }else{
                        log.error("请求的userId:"+toUserId+"不在该服务器上");
                        //否则不在这个服务器上，发送到mysql或者redis
                    }
                }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) throws IOException {
            this.session.getBasicRemote().sendText(message);
        }
    
    
        /**
         * 发送自定义消息
         * */
        public static void sendInfo(String message,@PathParam("userId") String userId) throws IOException {
            log.info("发送消息到:"+userId+"，报文:"+message);
            if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
                webSocketMap.get(userId).sendMessage(message);
            }else{
                log.error("用户"+userId+",不在线！");
            }
        }
    
        public static synchronized int getOnlineCount() {
            return onlineCount;
        }
    
        public static synchronized void addOnlineCount() {
            WebSocketServer.onlineCount++;
        }
    
        public static synchronized void subOnlineCount() {
            WebSocketServer.onlineCount--;
        }
    }

### 4.集群版 ###

在使用websocket的时候会面对session共享的问题

在这里我们通过Redis的监听订阅模式来实现session共享，当有连接加入时，将连接发送到队列中然后向订阅该队列的服务下发，由存储该连接的服务进行逻辑处理

session共享的业务逻辑如下图所示

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5Lq66Iu86Iul5Y-q6Iy55Yid6KaL_size_20_color_FFFFFF_t_70_g_se_x_16][]

#### [ 4.1Redis发布订阅模式实现][4.1Redis] ####

#### [Redis发布订阅模式实现原理][Redis] ####

#### 4.2队列消息推送 ####

@OnMessage
        public void onMessage(Session session, String message) throws IOException, ParseException {
            //业务逻辑处理
            //向所有用户广播
            redisUtil.convertSend("topic", message);
        }

#### 4.3队列消息接收 ####

/**
         * 队列消息接收方法
         * 2022/2/21
         */
        public void receiveMessage(String message) {
            log.info("[开始消费REDIS消息队列topic数据...]");
            try {
                //业务逻辑处理
                webSocketServer.sendMessageForProject(message);
                log.info("[消费REDIS消息队列topic数据成功.]");
            } catch (Exception e) {
                log.error("[消费REDIS消息队列topic数据失败，失败信息:{}]", e.getMessage());
            }
            latch.countDown();
        }

##  ##

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5Lq66Iu86Iul5Y-q6Iy55Yid6KaL_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/e7001e09490d46a1b80055c4a3e5d90c.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5Lq66Iu86Iul5Y-q6Iy55Yid6KaL,size_20,color_FFFFFF,t_70,g_se,x_16
[4.1Redis]: https://blog.csdn.net/printf88/article/details/122337464?spm=1001.2014.3001.5501
[Redis]: https://blog.csdn.net/printf88/article/details/123377831?spm=1001.2014.3001.5502