SpringBoot集成WebSocket，实现后台向前端实时推送信息

拼搏现实的明天。 2023-07-24 05:41 1阅读 0赞

##### 使用场景： #####

当前端调用WebSocket时，后台从第三方接口获取数据，实时推送到前端（每隔5秒）。

### 一、什么是WebSocket？ ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70]

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

### 二、为什么需要 WebSocket ？ ###

初次接触 WebSocket 的人，都会问同样的问题：我们已经有了 HTTP 协议，为什么还需要另一个协议？他能带来什么好处？

答案很简单，因为 HTTP 协议有一个缺陷：`通信只能由客户端发起`。

举例来说，我们想了解今天的天气，只能是客户端向服务器发出请求，服务器返回查询结果。HTTP 协议做不到服务器主动向客户端推送信息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 1]

这种单向请求的特点，注定了如果服务器有连续的状态变化，客户端要获知就非常麻烦。我们只能使用"[轮询][Link 1]"：每隔一段时候，就发出一个询问，了解服务器有没有新的信息。最典型的场景就是`聊天室`。  
轮询的效率低，非常浪费资源（因为必须不停连接，或者 HTTP 连接始终打开）。因此，工程师们一直在思考，有没有更好的方法。WebSocket 就是这样发明的。

### 三、WebSocket 简介 ###

WebSocket 协议在2008年诞生，2011年成为国际标准。所有浏览器都已经支持了。

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 2]

其他特点包括：

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

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

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

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

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

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

ws://127.0.0.1:80/ws/path

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 3]

### 四、代码实现 ###

#### 4.1 maven 依赖 ####

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

因涉及到 js 连接服务端，这里也写了调用 WebSocket 的 html，此处集成了 thymeleaf 模板。【前后分离的项目可省略，此处都是前端的工作。】

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

配置文件：

server:
      port: 8082
     
    #添加Thymeleaf配置
    thymeleaf:
      cache: false
      prefix: classpath:/templates/
      suffix: .html
      mode: HTML5
      encoding: UTF-8
      content-type: text/html

#### 4.2 启动 WebSocket 支持 ####

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    /** * 开启 WebSocket 支持 * @author Siona * @date 2020/4/8 17:40 **/
    @Configuration
    public class WebSocketConfig { 
    
        @Bean
        public ServerEndpointExporter serverEndpointExporter() { 
            return new ServerEndpointExporter();
        }
    
    }

#### 4.3 自定义 WebSocketServer ####

WebSocket 的核心代码。

（1）WebSocket 是类似客户端服务端的形式（采用 ws 协议），此处的 WebSocketServer相当于一个 `ws` 协议的 Controller。

（2）实现 `@OnOpen` 开启连接，`@onClose` 关闭连接，`@onMessage` 接收消息等方法。

import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.shingis.common.exception.ApiException;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.stereotype.Component;
    
    import javax.websocket.*;
    import javax.websocket.server.PathParam;
    import javax.websocket.server.ServerEndpoint;
    import java.io.IOException;
    import java.util.Iterator;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /** * @author Siona * @date 2020/4/8 17:43 **/
    @Slf4j
    @ServerEndpoint("/ws/{userId}")
    @Component
    public class WebSocketServer { 
    
        /** * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的 */
        private static int onlineCount = 0;
    
        /** * concurrent 包的线程安全Set，用来存放每个客户端对应的 myWebSocket对象 * 根据userId来获取对应的 WebSocket */
        private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    
        /** * 与某个客户端的连接会话，需要通过它来给客户端发送数据 */
        private Session session;
    
        /** * 接收 sid */
        private String userId = "";
    
    
        /** * 连接建立成功调用的方法 * * @param session * @param userId */
        @OnOpen
        public void onOpen(Session session, @PathParam("userId") String userId) { 
            this.session = session;
            this.userId = userId;
    
            webSocketMap.put(userId, this);
            log.info("webSocketMap -> " + JSON.toJSONString(webSocketMap));
    
            addOnlineCount(); // 在线数 +1
            log.info("有新窗口开始监听:" + userId + ",当前在线人数为" + getOnlineCount());
    
            try { 
                sendMessage(JSON.toJSONString("连接成功"));
            } catch (IOException e) { 
                e.printStackTrace();
                throw new ApiException("websocket IO异常！！！！");
            }
    
        }
    
        /** * 关闭连接 */
    
        @OnClose
        public void onClose() { 
            if (webSocketMap.get(this.userId) != null) { 
                webSocketMap.remove(this.userId);
                subOnlineCount(); // 人数 -1
                log.info("有一连接关闭，当前在线人数为：" + getOnlineCount());
            }
        }
    
        /** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 * @param session */
        @OnMessage
        public void onMessage(String message, Session session) { 
            log.info("收到来自窗口" + userId + "的信息：" + message);
    
            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.info("请求的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();
        }
    
        /** * 实现服务器主动推送 * * @param message * @throws IOException */
        public void sendMessage(String message) throws IOException { 
            this.session.getBasicRemote().sendText(message);
        }
    
        /** * 群发自定义消息 * * @param message * @param userId * @throws IOException */
        public static void sendInfo(String message, @PathParam("userId") String userId) throws IOException { 
    
            // 遍历集合，可设置为推送给指定sid，为 null 时发送给所有人
            Iterator entrys = webSocketMap.entrySet().iterator();
            while (entrys.hasNext()) { 
                Map.Entry entry = (Map.Entry) entrys.next();
    
                if (userId == null) { 
                    webSocketMap.get(entry.getKey()).sendMessage(message);
                    log.info("发送消息到：" + entry.getKey() + "，消息：" + message);
                } else if (entry.getKey().equals(userId)) { 
                    webSocketMap.get(entry.getKey()).sendMessage(message);
                    log.info("发送消息到：" + entry.getKey() + "，消息：" + message);
                }
    
            }
    
    
        }
    
        private static synchronized int getOnlineCount() { 
            return onlineCount;
        }
    
        private static synchronized void addOnlineCount() { 
            WebSocketServer.onlineCount++;
        }
    
        private static synchronized void subOnlineCount() { 
            WebSocketServer.onlineCount--;
        }
    }

#### 4.4 实时消息推送 ####

此处需求是 项目启动就会自动推送到前端，前端开启 WebSocket 后进行接收。  
调用两个第三方接口的数据 同时推送给前端，如果是前端点击不同的页面或按钮只需要一个接口的数据，则改为使用两个 WebSocket 分别推送即可。

import lombok.SneakyThrows;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.core.annotation.Order;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    
    /** * @author Siona * @date 2020/4/9 10:12 **/
    @Slf4j
    @Component  // 被Spring容器管理
    @Order(1)   // 如果多个自定义ApplicationRunner，用来表明执行顺序
    public class PushAlarm implements ApplicationRunner {    // 服务启动后自动加载该类
    
        @Autowired
        GasSupport gasSupport;
    
        @Override
        public void run(ApplicationArguments args) throws Exception { 
            log.info("------------->" + "项目启动，now =" + new Date());
            this.myTimer();
        }
    
        public void myTimer() { 
            
            String userId = null; // userId 为空时，会推送给连接此 WebSocket 的所有人
    
            Runnable runnable1 = new Runnable() { 
                @SneakyThrows
                @Override
                public void run() { 
                    while (true) { 
                        String message = gasSupport.GetWasteGasRealData(""); // 第三方接口返回数据
                        WebSocketServer.sendInfo(message, userId); // 推送
                        Thread.sleep(5000);
                    }
                }
            };
    
            Runnable runnable2 = new Runnable() { 
                @SneakyThrows
                @Override
                public void run() { 
                    while (true) { 
                        String message = gasSupport.GetWasteWaterRealData(""); // 第三方接口返回数据
                        WebSocketServer.sendInfo(message, userId); // 推送
                        Thread.sleep(5000);
                    }
                }
            };
    
            Thread thread1 = new Thread(runnable1);
            Thread thread2 = new Thread(runnable2);
    
            thread1.start();
            thread2.start();
            
        }
    
    }

#### 4.5 前端页面 ####

页面用 js 代码 `调用 WebSocket` ，最新的浏览器一般都支持（我用的谷歌浏览器）。最重要的一点就是使用 `ws` 协议，如果使用了一些路径类，可以用 `replace("http","ws")` 来替换。

<!DOCTYPE html>
    <html>
    <head>
        <meta charset="utf-8">
        <title>websocket通讯</title>
    </head>
    <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.js"></script>
    <script>
        var socket;
        function openSocket() { 
            if(typeof(WebSocket) == "undefined") { 
                console.log("您的浏览器不支持WebSocket");
            }else{ 
                console.log("您的浏览器支持WebSocket");
                //实现化WebSocket对象，指定要连接的服务器地址与端口 建立连接
                //等同于socket = new WebSocket("ws://localhost:8888/xxxx/im/25");
                //var socketUrl="${request.contextPath}/im/"+$("#userId").val();
                var socketUrl="http://localhost:5001/ws/"+$("#userId").val();
                socketUrl=socketUrl.replace("https","ws").replace("http","ws");
                console.log(socketUrl);
                if(socket!=null){ 
                    socket.close();
                    socket=null;
                }
                socket = new WebSocket(socketUrl);
                //打开事件
                socket.onopen = function() { 
                    console.log("websocket已打开");
                    //socket.send("这是来自客户端的消息" + location.href + new Date());
                };
                //获得消息事件
                socket.onmessage = function(msg) { 
                    console.log(msg.data);
                    //发现消息进入 开始处理前端触发逻辑
                };
                //关闭事件
                socket.onclose = function() { 
                    console.log("websocket已关闭");
                };
                //发生了错误事件
                socket.onerror = function() { 
                    console.log("websocket发生了错误");
                }
            }
        }
        function sendMessage() { 
            if(typeof(WebSocket) == "undefined") { 
                console.log("您的浏览器不支持WebSocket");
            }else { 
                console.log("您的浏览器支持WebSocket");
                console.log('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
                socket.send('{"toUserId":"'+$("#toUserId").val()+'","contentText":"'+$("#contentText").val()+'"}');
            }
        }
    </script>
    <body>
    <p>【userId】：<div><input id="userId" name="userId" type="text" value="10"></div>
    <p>【toUserId】：<div><input id="toUserId" name="toUserId" type="text" value="20"></div>
    <p>【toUserId】：<div><input id="contentText" name="contentText" type="text" value="hello websocket"></div>
    <p>【操作】：<div><a onclick="openSocket()">开启socket</a></div>
    <p>【操作】：<div><a onclick="sendMessage()">发送消息</a></div>
    </body>
    </html>

#### 4.6 编写 Controller 类 ####

跳转到指定页面（webSocket.html）

import org.springframework.http.ResponseEntity;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import org.springframework.web.servlet.ModelAndView;
    
    /** * @author Siona * @date 2020/4/8 18:40 **/
    @RestController
    public class DemoController { 
    
        @GetMapping("/index")
        public ResponseEntity<String> index() { 
            return ResponseEntity.ok("请求成功");
        }
    
        @GetMapping("/page")
        public ModelAndView page() { 
            return new ModelAndView("webSocket");
        }
    }

#### 4.7 运行 ####

浏览器输入 `http://localhost:5001/page` ，打开 webSocket.html 页面，F12 打开控制台查看测试结果。

开启socket后，可以看到控制台出现服务端不断推送的数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 4]

#### 小结 ####

ConcurrentHashMap：保证多线程安全，同时方便利用 `map.get(userId)` 进行推送到指定窗口。

相比Set，Set遍历是费事且麻烦的事情，而Map的get是简单便捷的，当WebSocket数量大的时候，这个小小的消耗就会聚少成多，影响体验，所以需要优化。在IM的场景下，指定userId进行推送消息更加方便。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70]: /images/20230528/6be5dd64cc41444bb2f78f38f8e6a998.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 1]: /images/20230528/411374f5eb0549c9826173809957851c.png
[Link 1]: https://www.pubnub.com/blog/http-long-polling/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 2]: /images/20230528/95048d49c55648e0a91e5c1cba513225.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 3]: /images/20230528/b848b90270e74d2f8399813357794d1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzODMzMzI3_size_16_color_FFFFFF_t_70 4]: /images/20230528/1f0e69f7e0c7444f96de7db002d85c04.png