springboot整合webscoket长连接（十）

阳光穿透心脏的1/2处 2023-10-07 15:53 17阅读 0赞

### http ###

http协议是用在应用层的协议，是基于tcp协议的，http协议建立链接也必须要有三次握手才能发送信息。  
http链接分为短链接，长链接，短链接是每次请求都要三次握手才能发送自己的信息。即每一个request对应一个response。长链接是在一定的期限内保持链接。保持TCP连接不断开。客户端与服务器通信，必须要有客户端发起然后服务器返回结果。客户端是主动的，服务器是被动的。

### WebSocket是什么 ###

WebSocket 是独立的、创建在 TCP 上的协议。
    WebSocket是一种在单个TCP连接上进行全双工通信的协议。
    WebSocket API也被W3C定为标准。
    WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。
    在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，
    并进行双向数据传输。
    Websocket 通过HTTP/1.1 协议的101状态码进行握手。
    为了创建Websocket连接，需要通过浏览器发出请求，之后服务器进行回应，
    这个过程通常称为“握手”（handshaking）。

### WebSocket可以用来做什么 ###

最早实现推送技术：轮询

很多网站为了实现推送技术，所用的技术都是轮询。
    轮询是在特定的的时间间隔（如每1秒），由浏览器对服务器发出HTTP请求，
    然后由服务器返回最新的数据给客户端的浏览器。这种传统的模式带来很明显的缺点，
    即浏览器需要不断的向服务器发出请求，然而HTTP请求可能包含较长的头部，
    其中真正有效的数据可能只是很小的一部分，显然这样会浪费很多的带宽等资源。
    而比较新的技术去做轮询的效果是Comet。这种技术虽然可以双向通信，但依然需要反复发出请求。
    而且在Comet中，普遍采用的长链接，也会消耗服务器资源。

在这种情况下，HTML5定义了WebSocket协议，能更好的节省服务器资源和带宽，并且能够更实时地进行通讯。浏览器通过 JavaScript 向服务器发出建立 WebSocket 连接的请求，连接建立以后，客户端和服务器端就可以通过 TCP 连接直接交换数据。![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw_size_16_color_FFFFFF_t_70]

参考链接: [HTML5 WebSocket][].

### WebSocket优点 ###

#### 较少的控制开销优点 ####

在连接创建后，服务器和客户端之间交换数据时，用于协议控制的数据包头部相对较小。
    在不包含扩展的情况下，对于服务器到客户端的内容，此头部大小只有2至10字节（和数据包长度有关）；
    对于客户端到服务器的内容，此头部还需要加上额外的4字节的掩码。
    相对于HTTP请求每次都要携带完整的头部，此项开销显著减少了。

#### 更强的实时性 ####

由于协议是全双工的，所以服务器可以随时主动给客户端下发数据。
    相对于HTTP请求需要等待客户端发起请求服务端才能响应，延迟明显更少；
    即使是和Comet等类似的长轮询比较，其也能在短时间内更多次地传递数据。

#### 保持连接状态 ####

与HTTP不同的是，Websocket需要先创建连接，这就使得其成为一种有状态的协议，
    之后通信时可以省略部分状态信息。而HTTP请求可能需要在每个请求都携带状态信息（如身份认证等）。
    更好的二进制支持。Websocket定义了二进制帧，相对HTTP，可以更轻松地处理二进制内容。

#### 可以支持扩展 ####

Websocket定义了扩展，用户可以扩展协议、实现部分自定义的子协议。
    如部分浏览器支持压缩等。

#### 更好的压缩效果 ####

相对于HTTP压缩，Websocket在适当的扩展支持下，可以沿用之前内容的上下文，
    在传递类似的数据时，可以显著地提高压缩率。

### 导入相应依赖 ###

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

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    /**
     * WebSocket配置
     */
    @Configuration
    public class WebSocketConfig {
        
    
        @Bean
        public ServerEndpointExporter serverEndpointExporter () {
        
            return new ServerEndpointExporter();
        }
    
    }

import lombok.extern.log4j.Log4j2;
    import org.springframework.stereotype.Component;
    
    import javax.websocket.*;
    import javax.websocket.server.ServerEndpoint;
    import java.io.IOException;
    import java.util.concurrent.CopyOnWriteArraySet;
    import java.util.concurrent.atomic.AtomicInteger;
    
    
    @Log4j2
    @Component
    @ServerEndpoint(value = "/ws")
    public class WebSocketServer {
        
    
        //与某个客户端的连接会话，需要通过它来给客户端发送数据
        private Session session;
    
        private static final AtomicInteger OnlineCount = new AtomicInteger(0);
    
        // concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
        private static CopyOnWriteArraySet<Session> SessionSet = new CopyOnWriteArraySet<Session>();
    
    
        /**
         * 连接建立成功调用的方法
         */
        @OnOpen
        public void onOpen(Session session) {
        
            SessionSet.add(session);
            this.session = session;
            int cnt = OnlineCount.incrementAndGet(); // 在线数加1
            log.info("有连接加入，当前连接数为：{}", cnt);
            SendMessage(session, "连接成功");
        }
    
        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose(Session session) {
        
            SessionSet.remove(session);
            int cnt = OnlineCount.decrementAndGet();
            log.info("有连接关闭，当前连接数为：{}", cnt);
        }
    
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message, Session session) {
        
            log.info("来自客户端的消息：{}", message);
            SendMessage(session, "收到消息，消息内容：" + message);
    
        }
    
        /**
         * 出现错误
         *
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
        
            log.error("发生错误：{}，Session ID： {}", error.getMessage(), session.getId());
            error.printStackTrace();
        }
    
        /**
         * 发送消息，实践表明，每次浏览器刷新，session会发生变化。
         *
         * @param session
         * @param message
         */
        public static void SendMessage(Session session, String message) {
        
            try {
        
                session.getBasicRemote().sendText(String.format("%s (From Server，Session ID=%s)", message, session.getId()));
            } catch (IOException e) {
        
                log.error("发送消息出错：{}", e.getMessage());
                e.printStackTrace();
            }
        }
    
        /**
         * 群发消息
         *
         * @param message
         * @throws IOException
         */
        public static void BroadCastInfo(String message) {
        
            for (Session session : SessionSet) {
        
                if (session.isOpen()) {
        
                    SendMessage(session, message);
                }
            }
        }
    
        /**
         * 指定Session发送消息
         *
         * @param sessionId
         * @param message
         * @throws IOException
         */
        public static void SendMessage(String message, String sessionId) {
        
            Session session = null;
            for (Session s : SessionSet) {
        
                if (s.getId().equals(sessionId)) {
        
                    session = s;
                    break;
                }
            }
            if (session != null) {
        
                SendMessage(session, message);
            } else {
        
                log.warn("没有找到你指定ID的会话：{}", sessionId);
            }
        }
    
    }

### webscoket在线测试网站 ###

链接: [http://coolaf.com/tool/chattest][http_coolaf.com_tool_chattest].

ws://127.0.01:8080/ws  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw_size_16_color_FFFFFF_t_70 1]

### 心跳机制及断线重连 ###

#### WebSocket 协议本质上是一个基于 TCP 的协议 ####

为了建立一个 WebSocket 连接，客户端浏览器首先要向服务器发起一个 HTTP 请求，这个请求和通常的 HTTP 请求不同，包含了一些附加头信息，其中附加头信息"Upgrade: WebSocket"表明这是一个申请协议升级的 HTTP 请求，服务器端解析这些附加的头信息然后产生应答信息返回给客户端，客户端和服务器端的 WebSocket 连接就建立起来了，双方就可以通过这个连接通道自由的传递信息，***并且这个连接会持续存在直到客户端或者服务器端的某一方主动的关闭连接***。这就导致了，客户端因意外掉线或者服务端意外宕机，双方却都无法感知到对方的在线状态，客户端无法及时重连或者服务端无法及时关闭无效连接。

#### 问题： ####

网络拓扑纷繁复杂、而从始节点A到终节点B之间可能要经过N多的交换机、路由器、防火墙等等硬件设备，每个硬件设备的相关设定也不统一，再加上网络中可能出现的拥塞、延迟等，使得我们在编程时，处理掉线也非常棘手，从程序的角度来说，我们可以总结为两种情况：程序能立即感知的掉线和程序不能立即感知的掉线。心跳机制就是用来解决这类问题。

#### 心跳机制的原理很简单 ####

客户端每隔N秒向服务端发送一个心跳消息，服务端收到心跳消息后，回复同样的心跳消息给客户端。如果服务端或客户端在M秒（M>N）内都没有收到包括心跳消息在内的任何消息，即心跳超时，我们就认为目标TCP连接已经断开了。

由于不同的应用程序对感知掉线的灵敏度不一样，所以，N和M的值就可以设定的不一样。灵敏度要求越高，N和M就要越小；灵敏度要求越低，N和M就可以越大。而要求灵敏度越高，也是有代价的，那就是需要更频繁地发送心跳消息，如果有几千个连接同时频繁地发送心跳消息，那么其所消耗的资源也是不能忽略的。

当然，网络环境（如延迟的大小）的好坏，也对会对N和M的值的设定产生影响，比如，网络延迟较大，那么N与M之间的差值也应该越大（比如，M是N的3倍）。否则，可能会产生误判 – 即连接没有断开，只是因为网络延迟大才及时没收到心跳消息，我们却认为TCP连接已经断开了。  
[参考连接][Link 1].

#### 前端代码 使用定时器来实现心跳 和 断线重连 ####

<!DOCTYPE HTML>
    <html>
       <head>
       <meta charset="utf-8">
       <title>菜鸟教程(runoob.com)</title>
        
          <script type="text/javascript">	  
    			
    			 var ws = null;
    			 var interval = null;
    			 var uri = "ws://192.168.252.1:8080/ws";
    			 var myVar = null;
    			 var reconnectState = null;
    			 
    			 function WebSocketTest(){
        
    				if ("WebSocket" in window){
        
    				   createWebSocket();
    				}else{
        
    				   // 浏览器不支持 WebSocket
    				   alert("您的浏览器不支持 WebSocket!");
    				}
    			 }
    		 
    		 	/**
    			 * 创建websocket
    			 */
    			function createWebSocket() {
        
    				alert("开启长连接")
                   // 打开一个 web socket
                   ws = new WebSocket(uri);
                   ws.onopen = function(){
        
    				  document.getElementById("reconnectShow").innerText= "无";
    				  document.getElementById("online").innerText= "在线";
    				  reconnectState = false;
    				  clearInterval(myVar);
    				  // 心跳检测
    				  interval = setInterval(function () {
        
    					// 使用 send() 方法发送数据		
    					sendMsg("ping");
    				  }, 5000);
    				  
                   };
                    
                   ws.onmessage = function (evt){
         
    					var received_msg = evt.data;
    					if(received_msg == "pong"){
        
    						console.log(received_msg);
    					}else{
        
    						document.getElementById("num").innerText= received_msg;
    					}
                   };
                    
                   ws.onclose = function(){
         
    				 webscoketClose();
                   };   
    			}
    			
    			
    		 	/**
    			 * websocket重连
    			 */
    			function reconnect() {
        
    				// 检测到websocket连接断开
    				// 0 (WebSocket.CONNECTING)正在链接中
    				// 1 (WebSocket.OPEN)已经链接并且可以通讯
    				// 2 (WebSocket.CLOSING) 连接正在关闭
    				// 3 (WebSocket.CLOSED)连接已关闭或者没有链接成功
    					if(reconnectState == false){
        
    						reconnectState = true;
    						var reconnectNum = 1;
    						console.log(ws.readyState);
    						myVar = setInterval(function(){
        
    			　　　　　　　　if(ws.readyState == 2 || ws.readyState == 3){
        
    			　　　　　　　　　　if(reconnectNum == 6){
        
    			　　　　　　　　　　　　clearInterval(myVar);
    									document.getElementById("reconnectShow").innerText= "停止重连";
    			　　　　　　　　　　　　return;
    			　　　　　　　　　　}
    								document.getElementById("reconnectShow").innerText= "第" + reconnectNum + "次尝试重连中";
    								createWebSocket();
    			　　　　　　　　　　reconnectNum ++;
    			　　　　　　　　}
    			
    			　　　　　　},5*1000)
    					}
    			}
    			
    			function noReconnectclose() {
        
    				reconnectState = true;
    				webscoketClose();
    			}
    			
    			/**
    			 * websocket关闭 
    			 */
    			function webscoketClose() {
        
    				ws.close();
    				clearInterval(interval)
    				console.log("websocket关闭");
    				document.getElementById("online").innerText= "离线";
    				// 开启断线重连
    				reconnect();
    			}
    			
    			function sendMsg(str){
        
    				if(ws.readyState == 1){
        
    					ws.send(str);
    				}else{
        
    					alert("未连接上服务器");
    				}
    			}
    			
    			function add() {
        	
    				sendMsg("+1");
    			}
    			
    			function remove() {
        
    				sendMsg("-1");
    			}
    					
          </script>
            
       </head>
       <body>
            <div id="sse">
             <a href="javascript:WebSocketTest()">运行 WebSocket</a>
    		 <a href="javascript:noReconnectclose()">停止 WebSocket</a>
            </div>
    			<br>
    			当前webscoket状态：<span id="online">离线</span>
    			<br>
    			当前重连状态：<span id="reconnectShow">无</span>
    			<br>
    			当前怒气值：<span id="num">0</span>
    		<br>
    		<a href="javascript:add()">+1</a>
    		<a href="javascript:remove()">-1</a>
       </body>
    </html>

#### 后端代码也是使用定时器来实现心跳 ####

import lombok.extern.log4j.Log4j2;
    import org.springframework.stereotype.Component;
    
    import javax.websocket.*;
    import javax.websocket.server.ServerEndpoint;
    import java.io.IOException;
    import java.util.Timer;
    import java.util.TimerTask;
    import java.util.concurrent.CopyOnWriteArraySet;
    import java.util.concurrent.atomic.AtomicInteger;
    
    
    @Log4j2
    @Component
    @ServerEndpoint(value = "/ws")
    public class WebSocketServer {
        
    
        //与某个客户端的连接会话，需要通过它来给客户端发送数据
        private Session session;
    
        private static final AtomicInteger OnlineCount = new AtomicInteger(0);
    
        // concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
        private static CopyOnWriteArraySet<Session> SessionSet = new CopyOnWriteArraySet<Session>();
    
        private static int num = 0;
    
        private Timer timer = new Timer();
    
        /**
         * 连接建立成功调用的方法
         */
        @OnOpen
        public void onOpen(Session session) {
        
            SessionSet.add(session);
            this.session = session;
            int cnt = OnlineCount.incrementAndGet(); // 在线数加1
            log.info("有连接加入，当前连接数为：{}", cnt);
            SendMessage(this.session, num+"");
            heartBeat(session);
        }
    
        /**
         * 连接关闭调用的方法
         */
        @OnClose
        public void onClose() {
        
            SessionSet.remove(this.session);
            int cnt = OnlineCount.decrementAndGet();
            log.info("有连接关闭，当前连接数为：{}", cnt);
        }
    
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message 客户端发送过来的消息
         */
        @OnMessage
        public void onMessage(String message, Session session) {
        
            log.info("收到客户端消息："+message);
            if("ping".equalsIgnoreCase(message)){
        
                SendMessage(session, "pong" );
                timer.cancel();
                heartBeat(session);
                return;
            }
            if("+1".equalsIgnoreCase(message)){
        
                this.num ++;
            }
            if("-1".equalsIgnoreCase(message)){
        
                this.num --;
            }
            BroadCastInfo(""+num );
        }
    
        /**
         * 出现错误
         *
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
        
            log.error("发生错误：{}，Session ID： {}", error.getMessage(), session.getId());
            error.printStackTrace();
        }
    
        /**
         * 心跳
         * @param session
         */
        private void heartBeat(Session session) {
        
            timer = new Timer();
            timer.schedule(new TimerTask() {
        
                @Override
                public void run() {
        
                    try {
        
                        session.close();
                    } catch (IOException e) {
        
                        e.printStackTrace();
                    }
                }
            }, 40000);
        }
    
        /**
         * 发送消息，实践表明，每次浏览器刷新，session会发生变化。
         *
         * @param session
         * @param message
         */
        public static void SendMessage(Session session, String message) {
        
            try {
        
                if (session.isOpen()) {
        
                    session.getBasicRemote().sendText(message);
                }
            } catch (IOException e) {
        
                log.error("发送消息出错：{}", e.getMessage());
                e.printStackTrace();
            }
        }
    
        /**
         * 群发消息
         *
         * @param message
         * @throws IOException
         */
        public static void BroadCastInfo(String message) {
        
            for (Session session : SessionSet) {
        
                SendMessage(session, message);
            }
        }
    
        /**
         * 指定Session发送消息
         *
         * @param sessionId
         * @param message
         * @throws IOException
         */
        public static void SendMessage(String message, String sessionId) {
        
            Session session = null;
            for (Session s : SessionSet) {
        
                if (s.getId().equals(sessionId)) {
        
                    session = s;
                    break;
                }
            }
            if (session != null) {
        
                SendMessage(session, message);
            } else {
        
                log.warn("没有找到你指定ID的会话：{}", sessionId);
            }
        }
    
    }

### getBasicRemote 和 getAsyncRemote 方法及发送消息注意事项 ###

session.getAsyncRemote().sendText(message); 是非阻塞式的

session.getBasicRemote().sendText(message, false); 是阻塞式的，一次发送全部消息

session.getBasicRemote().sendText(message, true)是阻塞式的，可以发送消息中的部分消息

交互数据不能过大，尽量小

### 如何获取HttpSession认证用户及开启连接带上参数 ###

参考链接: [基于WebSocket的在线聊天室][WebSocket].

### Websocket不能注入bean的解决办法 ###

一、问题现象  
已用@component标注，使用Atuowire注入bean时失败。bean等于null。

二、问题原因  
spring维护的是单实例，websocket为多实例。每次请求建立链接的时候，都会生成一个新的bean。

三、解决方案  
让websoctserver持有一个静态的bean对象，在创建时通过@Atuowire标注方法，自动注入值。

四、示意代码

private static Service service;
    
    @Autowired
    public void setService(Service service) {
        
        this.service = service;
    }

### 踩坑注意 ###

链接: [Java实现websocket发送长文本问题][Java_websocket]

Java实现websocket时，每次发送的文本长度默认限制在8k范围内

如果发送超过8k的消息，就会报错：  
No async message support and buffer too small Buffer size: \[8,192\], Message size: \[10,240\]

#### 1 实现ServletContextInitializer接口，对onStartup进行重载 （全局） ####

@Configuration
    @ComponentScan
    @EnableAutoConfiguration
    public class WebAppRootContext implements ServletContextInitializer {
        
    
        //设置servlet 字符串 大小设置 解决 分析范围数据太大 造成的连接中断问题
        @Override
        public void onStartup(ServletContext servletContext) throws ServletException {
        
            System.out.println("org.apache.tomcat.websocket.textBufferSize");
            servletContext.addListener(WebAppRootListener.class);
            servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize","10485760");// 单位是b
        }
    
    }

但是这样**每个websocket会话初始化**就会占用设置的内存，这样连接达到一定数量就会内存溢出。该操作慎用，很容易造成生产环境内存溢出。

#### 2 注解 @OnMessage设置消息大小 （局部） ####

// 当接收到消息时才分配缓冲空间，只在本webscoket类生效
        @OnMessage(maxMessageSize=5242880)
        public void onMessage(String message,  Session session) {
        
    
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210313154654494.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw,size_16,color_FFFFFF,t_70
[HTML5 WebSocket]: https://www.runoob.com/html/html5-websocket.html
[http_coolaf.com_tool_chattest]: http://coolaf.com/tool/chattest
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210313153417322.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNjA0ODkw,size_16,color_FFFFFF,t_70
[Link 1]: http://blog.oraycn.com/ESFramework_07.aspx
[WebSocket]: https://www.jianshu.com/p/62790429acef
[Java_websocket]: https://blog.csdn.net/asartear/article/details/128784844