SSE 和 WebSocket 应用

我不是女神ヾ 2023-10-15 16:27 36阅读 0赞

#### SSE 和 WebSocket 应用 ####

*  一.SSE 和 WebSocket 对比
 *  二.SSE 和 WebSocket 调试
 *   *  SpringBoot 下 SSE 应用
     *   *  1.依赖
         *  2.启动类
         *  3.接口类
         *  4.Html 测试
         *  5.测试结果
     *  SpringBoot 下 WebSocket 应用
     *   *  1.依赖
         *  2.启动类
         *  3.WS 切点配置
         *  4.WS连接类配置
         *  5.WS Html 测试
         *  6.测试结果

## 一.SSE 和 WebSocket 对比 ##

SSE 全称 Server-Send Events 基于 HTTP 的单向通信协议  
WebSocket 基于 HTTP 封装的 WS 双向通信协议

## 二.SSE 和 WebSocket 调试 ##

JVM 版本

![在这里插入图片描述][cc3b91f192e3444a94777d7228b74929.png]

### SpringBoot 下 SSE 应用 ###

#### 1.依赖 ####

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>socket-demo</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <maven.compiler.source>20</maven.compiler.source>
            <maven.compiler.target>20</maven.compiler.target>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <spring.version>3.1.3</spring.version>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.28</version>
            </dependency>
    
        </dependencies>
    
    
    </project>

#### 2.启动类 ####

package org.example;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    /**
     * @author Administrator
     */
    @SpringBootApplication
    public class SocketDemoApp {
        
        public static void main(String[] args) {
        
            SpringApplication.run(SocketDemoApp.class,args);
        }
    }

#### 3.接口类 ####

package org.example.controller;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.http.MediaType;
    import org.springframework.web.bind.annotation.CrossOrigin;
    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.mvc.method.annotation.SseEmitter;
    
    import java.io.IOException;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.atomic.AtomicReference;
    
    /**
     * @author zhuwd && moon
     * @Description
     * @create 2023-09-04 22:53
     */
    @Slf4j
    @CrossOrigin
    @RestController
    @RequestMapping("/test")
    public class TestController {
        
    
        /**
         * 线程副本变量
         */
        private ThreadLocal<AtomicReference<Boolean>> isSendThreadLocal = new ThreadLocal<>();
    
        /**
         * 缓存线程
         */
        private Map<Integer,AtomicReference<Boolean>> sidThreadMap = new ConcurrentHashMap<>();
    
        /**
         * Server Send Event
         *
         * @param type 1 温度 2 湿度
         * @return
         */
        @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
        public synchronized SseEmitter sendToClient(int sid,int type){
        
            SseEmitter emitter = new SseEmitter(1000000L);
            // 模拟生成实时股票价格并推送给客户端
            Random random = new Random();
            new Thread(() -> {
        
                try {
        
                    //缓存当前线程
                    AtomicReference temp = sidThreadMap.remove(sid);
                    if (null != temp){
        
                        temp.set(false);
                    }
                    //添加缓存
                    sidThreadMap.put(sid,new AtomicReference<>(false));
                    //缓存状态
                    isSendThreadLocal.set(sidThreadMap.get(sid));
                    //发数
                    while (true) {
        
                        try {
        
                            // 生成随机值
                            double val = 10 + random.nextDouble() * 10;
                            // 获取单位
                            String unit = type == 1 ? "℃":(type == 2?"%":"");
                            // 构造股票价格的消息
                            String message = String.format("%.2f", val) + unit;
                            // 发送消息给客户端
                            emitter.send(SseEmitter.event().data(message));
                            // 休眠 1 秒钟
                            Thread.sleep(1000);
                        } catch (IOException e) {
        
                            throw new RuntimeException(e);
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        } finally {
        
                            //判断是否退出
                            if (isSendThreadLocal.get().get()){
        
                                isSendThreadLocal.remove();
                                break;
                            }
                        }
                    }
                } catch (Exception e) {
        
                    emitter.completeWithError(e);
                }
            }).start();
            //返回
            return emitter;
        }
    
    
        /**
         * Stop Server Send Event
         * @param sid
         */
        @GetMapping("/stop")
        public void stop(int sid){
        
            //取出线程
            AtomicReference<Boolean> temp = sidThreadMap.remove(sid);
            if (null != temp){
        
                temp.set(true);
            }
        }
    
    }

#### 4.Html 测试 ####

<!DOCTYPE html>
    <html lang="en">
    <head>
      <meta charset="UTF-8">
      <title>温湿度监控</title>
    </head>
    <body>
    <h1>温度</h1>
    <h3><div id="temperature"></div></h3>
    <h1>湿度</h1>
    <h3><div id="humidity"></div></h3>
    
    <br/>
    <button onclick="stop(1,'temperature')">停止温度监控</button>
    <button onclick="stop(2,'humidity')">停止湿度监控</button>
    <button onclick="start(1)">恢复温度监控</button>
    <button onclick="start(2)">恢复湿度监控</button>
    
    
    
    <script src="https://code.jquery.com/jquery-3.6.1.js"></script>
    <script>
    
      const map = new Map();
    
      //EventSource 事件 onopen/onmessage/onerror
    
      function getSSE(sid,type,id){
          
        obj = new EventSource('http://127.0.0.1:8080/test/sse?sid=' + sid + '&type=' + type);
        obj.onmessage = function (event) {
          
          document.getElementById(id).innerHTML = event.data;
        };
    
        document.getElementById(id).style.color = "#00FA9A";//"#ff0000";
    
        obj.onopen = function (){
          
          console.log('obj opopen connect obj state:' + obj.readyState)
        }
    
        obj.onerror = function (){
          
          obj.close();
          console.log('obj onerror connect obj state:' + obj.readyState)
        }
    
        map.set(sid,obj)
      }
    
      function init() {
          
        getSSE(1,1,'temperature')
        getSSE(2,2,'humidity')
      }
    
      init();
    
      /**
       * 停止
       */
      function stop(sid,id){
          
        $.get("http://127.0.0.1:8080/test/stop?sid=" + sid,function(data,status){
          
          console.log("Data: " + data + "\nStatus: " + status);
          document.getElementById(id).style.color = "#ff0000";
          map.get(sid).close();
        });
      }
    
      /**
       * 开始
       */
      function start(sid){
          
        status = map.get(sid).readyState
        if (status == 1) {
          
          console.log('see is on connected...' + sid + ' status ' + status)
          return
        }
    
    
        if (sid == 1) {
          
          getSSE(1,1,'temperature')
        } else if (sid == 2) {
          
          getSSE(2,2,'humidity')
        }
      }
    
    </script>
    </body>
    </html>

#### 5.测试结果 ####

![在这里插入图片描述][4c39d92c7dbe499ca1a6f0c24fccdcae.png]

### SpringBoot 下 WebSocket 应用 ###

#### 1.依赖 ####

#### 2.启动类 ####

#### 3.WS 切点配置 ####

package org.example.config;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.web.socket.server.standard.ServerEndpointExporter;
    
    /**
     * @author zhuwd && moon
     * @Description
     * @create 2023-09-04 23:15
     */
    @Configuration
    public class WebSocketConfig {
        
    
        /**
         * 	注入 ServerEndpointExporter
         * 	这个 Bean 会自动注册使用了 @ServerEndpoint 声明的 Websocket Endpoint
         */
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
        
            return new ServerEndpointExporter();
        }
    }

#### 4.WS连接类配置 ####

package org.example.config;
    
    import jakarta.websocket.*;
    import jakarta.websocket.server.PathParam;
    import jakarta.websocket.server.ServerEndpoint;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.CopyOnWriteArraySet;
    
    /**
     * @author zhuwd && moon
     *
     * 接口路径 ws://localhost:8080/webSocket/{sid};
     *
     * @Description
     * @create 2023-09-04 23:16
     */
    
    @Slf4j
    @Component
    @ServerEndpoint("/websocket/{sid}")
    public class MyWebSocket {
        
    
        /**
         * ws 会话
         */
        private Session session;
    
        /**
         * 连接 id
         */
        private String sid;
    
        /**
         * 缓存类对象
         */
        private static CopyOnWriteArraySet<MyWebSocket> webSockets = new CopyOnWriteArraySet<>();
    
        /**
         * 缓存用户信息
         */
        private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<>();
    
        /**
         * 缓存用户信息
         */
        private static ConcurrentHashMap<Session,String> sidPool = new ConcurrentHashMap<>();
    
        /**
         * 链接成功调用的方法
         * @param session
         * @param sid
         */
        @OnOpen
        public void onOpen(Session session, @PathParam(value="sid") String sid) {
        
            try {
        
                this.session = session;
                this.sid = sid;
                webSockets.add(this);
                sessionPool.put(sid, session);
                sidPool.put(session,sid);
                log.info("one client join in sid {} all counts {}",sid,webSockets.size());
            } catch (Exception e) {
        
                log.error("join in error:",e);
            }
        }
    
        /**
         * 链接关闭调用的方法
         */
        @OnClose
        public void onClose() {
        
            try {
        
                webSockets.remove(this);
                sessionPool.remove(this.sid);
                log.info("one client leave sid {} all counts {}",sid,webSockets.size());
            } catch (Exception e) {
        
                log.error("leave error:",e);
            }
        }
    
        /**
         * 收到客户端消息后调用的方法
         *
         * @param message
         * @param session
         */
        @OnMessage
        public void onMessage(Session session,String message) {
        
            //发送
            sendOneMessage(sid,"Hello " + message);
            log.info("client sid {} message : {}",sidPool.get(session),message);
        }
    
        /** 发送错误时的处理
         * @param session
         * @param error
         */
        @OnError
        public void onError(Session session, Throwable error) {
        
            log.error("ws sid {} error:",sidPool.get(session),error);
        }
    
        /**
         * 广播
         * @param message
         */
        public void sendAllMessage(String message) {
        
            log.info("broadcast : {}",message);
            for(MyWebSocket webSocket : webSockets) {
        
                try {
        
                    if(webSocket.session.isOpen()) {
        
                       webSocket.session.getAsyncRemote().sendText(message);
                    }
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        /**
         * 单点发送
         * @param sid
         * @param message
         */
        public void sendOneMessage(String sid, String message) {
        
            Session session = sessionPool.get(sid);
            if (session != null && session.isOpen()) {
        
                try {
        
                    log.info("to one : {}",message);
                    session.getAsyncRemote().sendText(message);
                } catch (Exception e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        /**
         * 多点发送
         * @param sids
         * @param message
         */
        public void sendMoreMessage(String[] sids, String message) {
        
            for(String sid:sids) {
        
                Session session = sessionPool.get(sid);
                if (session != null&&session.isOpen()) {
        
                    try {
        
                        log.info("to more : {}",message);
                        session.getAsyncRemote().sendText(message);
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }
            }
        }
    }

#### 5.WS Html 测试 ####

<!DOCTYPE html>
    <html lang="en">
    <head>
      <meta charset="UTF-8">
      <title>WS</title>
    </head>
    <body>
    <h1>Message</h1>
    <h3><div><input id="msg" type="text" /></div></h3>
    <h1>Receive</h1>
    <h3><div><input id="rsg" type="text" readonly = true/></div></h3>
    <br/>
    
    <button onclick="send()">发送</button>
    
    
    <script src="https://code.jquery.com/jquery-3.6.1.js"></script>
    <script>
    
      var socket = new WebSocket('ws://127.0.0.1:8080/websocket/1');
    
      socket.onopen = function(evt){
          
    
      };
    
      socket.onerror = function(evt){
          
    
      };
    
      socket.onmessage = function(evt){
          
        console.log('------' + evt.data)
        $('#rsg').val(evt.data)
      };
    
      socket.onclose = function(evt){
          
    
      };
    
      function send(){
          
        socket.send($('#msg').val());
      }
    
    </script>
    </body>
    </html>

#### 6.测试结果 ####

![在这里插入图片描述][0cb1cff76d8c4ed2bdaf5696634a299a.png]

[cc3b91f192e3444a94777d7228b74929.png]: https://img-blog.csdnimg.cn/cc3b91f192e3444a94777d7228b74929.png
[4c39d92c7dbe499ca1a6f0c24fccdcae.png]: https://img-blog.csdnimg.cn/4c39d92c7dbe499ca1a6f0c24fccdcae.png
[0cb1cff76d8c4ed2bdaf5696634a299a.png]: https://img-blog.csdnimg.cn/0cb1cff76d8c4ed2bdaf5696634a299a.png