WebSocket 实现服务器消息推送客户端

客官°小女子只卖身不卖艺 2023-06-24 12:29 25阅读 0赞

### 一、背景 ###

项目需要做一个消息能够实时获取的功能，系统日活跃量达到10000，产生的消息是活跃量的数倍，如果采用 Http 的方式轮询后端服务，会使得后端服务压力过大而奔溃，因此需要一种新的技术方式来改变 “拉” 的方式。

### 二、解决方案 ###

经过各种 Google、百度 后发现可以使用 html5 的新技术 WebSocket ，将现有 “拉”消息的方式改变成 “推” 的模式，大大的减少服务器压力。

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

### 三、具体实现 ###

实例采用 Spring Boot 框架，

1.  引入 pom 依赖

<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-websocket</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-undertow</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    			<exclusions>
    				<exclusion>
    					<groupId>org.springframework.boot</groupId>
    					<artifactId>spring-boot-starter-tomcat</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>

1.  WebSocket 服务可采用 `websocket-api` 或 `spring-websocket` 开发，我们采用 `websocket-api` 的注解开发方式：

package com.gridsum.techpub.systemhistory.api.server;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Service;
    
    import javax.websocket.*;
    import javax.websocket.server.PathParam;
    import javax.websocket.server.ServerEndpoint;
    import java.io.IOException;
    import java.util.Objects;
    import java.util.Set;
    import java.util.concurrent.CopyOnWriteArraySet;
    
    /**
     * @author ouyangrongtao
     * @version 1.0
     * @description WebSocketServer
     * @date 2019/12/23 10:16
     **/
    @ServerEndpoint("/websocket/{sid}")
    @Service
    public class WebSocketServer {
    
        private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class);
    
        private ClientInfo clientInfo;
        /**
         * 存放每个客户端对应的 ClientInfo 对象。
         */
        private static final Set<ClientInfo> WEB_SOCKET_SET = new CopyOnWriteArraySet<>();
    
        /**
         * 连接建立成功调用的方法
         *
         * @param session 会话
         * @param sid     客户端
         */
        @OnOpen
        public void onOpen(Session session, @PathParam("sid") String sid) {
            //加入set中
            this.clientInfo = new ClientInfo(sid, session);
            WEB_SOCKET_SET.add(clientInfo);
            logger.info("有新窗口开始监听:[{}],当前在线人数为[{}]", sid, WEB_SOCKET_SET.size());
            try {
                this.sendMessage(session, "连接成功");
            } catch (IOException e) {
                logger.error("websocket IO异常");
            }
        }
    
        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose() {
            //从set中删除
            WEB_SOCKET_SET.remove(this.clientInfo);
            logger.info("有一连接关闭！当前在线人数为：[{}]", WEB_SOCKET_SET.size());
        }
    
        /**
         * 群发
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message) {
            logger.info("收到来自窗口[{}]的信息:[{}]", this.clientInfo.getSid(), message);
            //群发消息
            for (ClientInfo item : WEB_SOCKET_SET) {
                try {
                    this.sendMessage(item.getSession(), message);
                } catch (IOException ignored) {
                }
            }
        }
    
        /**
         * 错误时调用
         * @param session 会话
         * @param error   错误信息
         */
        @OnError
        public void onError(Session session, Throwable error) {
            logger.error("发生错误", error);
        }
    
        /**
         * 给 sid 发送消息
         * @param message 消息
         * @param sid sid
         */
        public void sendMessage(String message, String sid) {
            logger.info("推送消息到窗口[{}]，推送内容:[{}]", sid, message);
    
            ClientInfo client = WEB_SOCKET_SET.parallelStream()
                    .filter(item -> item.getSid().equals(sid)).findFirst().orElse(null);
            if (client != null) {
                try {
                    this.sendMessage(client.getSession(), message);
                } catch (IOException ignored) {
                }
            }
        }
    
        /**
         * 实现服务器主动推送
         * @param session session
         * @param message message
         * @throws IOException IOException
         */
        private void sendMessage(Session session, String message) throws IOException {
            session.getBasicRemote().sendText(message);
        }
    
    
        class ClientInfo {
            /**
             * 接收sid
             */
            private String sid = "";
    
            /**
             * 客户端
             */
            private Session session;
    
            public ClientInfo() { }
    
            private ClientInfo(String sid, Session session) {
                this.sid = sid;
                this.session = session;
            }
    
            private String getSid() {
                return sid;
            }
    
            private Session getSession() {
                return session;
            }
    
            @Override
            public boolean equals(Object o) {
                if (this == o) {
                    return true;
                }
                if (o == null || getClass() != o.getClass()) {
                    return false;
                }
                ClientInfo that = (ClientInfo) o;
                return Objects.equals(sid, that.sid);
            }
    
            @Override
            public int hashCode() {
                return Objects.hash(sid);
            }
        }
    }

1.  前端代码

<!DOCTYPE HTML>
    <html>
       <head>
       <meta charset="utf-8">
       <title>WebSocketClient1000001</title>
    
          <script type="text/javascript">
             function WebSocketTest() {
                if ("WebSocket" in window) {
                   // 打开一个 web socket
                   var ws = new WebSocket("ws://127.0.0.1:8080/websocket/1000001");
    
                   ws.onopen = function() {
                      // Web Socket 已连接上，使用 send() 方法发送数据
                      console.log("数据发送中...");
                   };
    
                   ws.onmessage = function (evt) { 
                      var received_msg = evt.data;
                      console.log("接收到消息：" + received_msg);
                   };
    
                   ws.onclose = function() { 
                      // 关闭 websocket
                      console.log("连接已关闭..."); 
                   };
                }
                else {
                   // 浏览器不支持 WebSocket
                   console.log("您的浏览器不支持 WebSocket!");
                }
             }
          </script>
    
       </head>
       <body>
    
          <div id="sse">
             <a href="javascript:WebSocketTest()">运行 WebSocketClient1000001</a>
          </div>
    
       </body>
    </html>

1.  来一个发消息的接口

/**
     * 发送消息给客户端
     * @author ouyangrongtao
     */
    @RestController
    public class WebSocketController {
    
        private WebSocketServer webSocketServer;
    
        @Autowired
        public WebSocketController(WebSocketServer webSocketServer) {
            this.webSocketServer = webSocketServer;
        }
    
        @PostMapping("/socket/push")
        public boolean pushToWeb(@RequestBody Map<String, String> content) {
            webSocketServer.sendMessage(content.get("message"), content.get("cid"));
            return true;
        }
    }

到此已经基本写完。使用 Postman 调用发消息的接口，发现客户端可以收到发送的消息。

### 四、问题记录 ###

在做的时候，因为项目用的 Tomcat 容器，导致 Tomcat 相关包与 WebSocket 依赖有冲突，最终项目不能启动，解决方式只需要将 Tomcat 容器改为 Undertow 。

<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-undertow</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    			<exclusions>
    				<exclusion>
    					<groupId>org.springframework.boot</groupId>
    					<artifactId>spring-boot-starter-tomcat</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>

异常信息：

Caused by: java.lang.IllegalStateException: javax.websocket.server.ServerContainer not available
    	at org.springframework.util.Assert.state(Assert.java:73)
    	at org.springframework.web.socket.server.standard.ServerEndpointExporter.afterPropertiesSet(ServerEndpointExporter.java:106)
    	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1753)
    	at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1690)
    	... 16 common frames omitted

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0NTk4NjAx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019122316193523.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0NTk4NjAx,size_16,color_FFFFFF,t_70