springboot整合sse

约定不等于承诺〃 2023-10-07 21:09 33阅读 0赞

链接: [SpringBoot 实现SSE 服务器发送事件][SpringBoot _SSE]  
链接: [SpringBoot 实现SSE 服务器发送事件][SpringBoot _SSE 1]  
链接: [Springboot之整合SSE实现消息推送][Springboot_SSE]  
链接: [springboot SseEmitter 消息推送][springboot SseEmitter]  
链接: [在springboot中使用Sse(Server-sent Events)Web实时通信技术-服务器发送事件SseEmitter][springboot_Sse_Server-sent Events_Web_-_SseEmitter]  
链接: [Web 实时消息推送详解][Web]

链接: [SSE（Server-Send Events）实战][SSE_Server-Send Events]

链接: [Server side event (SSE)实现消息推送功能][Server side event _SSE]

### 什么是sse ###

链接: [著作权归所有原文链接][Link 1]  
服务器发送事件(Server-Sent Events)，简称 SSE。这是一种服务器端到客户端(浏览器)的单向消息推送。SSE 基于 HTTP 协议的，我们知道一般意义上的 HTTP 协议是无法做到服务端主动向客户端推送消息的，但 SSE 是个例外，它变换了一种思路。

SSE 在服务器和客户端之间打开一个单向通道，服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息，在有数据变更时从服务器流式传输到客户端。整体的实现思路有点类似于在线视频播放，视频流会连续不断的推送到浏览器，你也可以理解成，客户端在完成一次用时很长（网络不畅）的下载。

SSE 与 WebSocket 作用相似，都可以建立服务端与浏览器之间的通信，实现服务端向客户端推送消息，但还是有些许不同：

*  SSE 是基于 HTTP 协议的，它们不需要特殊的协议或服务器实现即可工作；WebSocket 需单独服务器来处理协议。
 *  SSE 单向通信，只能由服务端向客户端单向通信；  
    WebSocket 全双工通信，即通信的双方可以同时发送和接受信息。
 *  SSE 实现简单开发成本低，无需引入其他组件；  
    WebSocket 传输数据需做二次解析，开发门槛高一些。
 *  SSE 默认支持断线重连；WebSocket 则需要自己实现。
 *  SSE 只能传送文本消息，二进制数据需要经过编码后传送；WebSocket 默认支持传送二进制数据。

***注意： SSE 不支持 IE 浏览器，对其他主流浏览器兼容性做的还不错。***

### SSE 与 WebSocket 该如何选择？ ###

链接: [著作权归所有原文链接][Link 1]  
SSE 好像一直不被大家所熟知，一部分原因是出现了 WebSocket，这个提供了更丰富的协议来执行双向、全双工通信。对于游戏、即时通信以及需要双向近乎实时更新的场景，拥有双向通道更具吸引力。但是，在某些情况下，不需要从客户端发送数据。而你只需要一些服务器操作的更新。比如：站内信、未读消息数、状态更新、股票行情、监控数量等场景，SEE 不管是从实现的难易和成本上都更加有优势。此外，SSE 具有 WebSocket 在设计上缺乏的多种功能，例如：自动重新连接、事件 ID 和发送任意事件的能力。

### sse 规范 ###

在 html5 的定义中，服务端 sse，一般需要遵循以下要求

*  请求头

开启长连接 + 流方式传递
    Content-Type: text/event-stream;charset=UTF-8
    Cache-Control: no-cache
    Connection: keep-alive

*  数据格式

服务端发送的消息，由 message 组成，其格式如下:

field:value\n\n

其中 field 有五种可能

*  空: 即以:开头，表示注释，可以理解为服务端向客户端发送的心跳，确保连接不中断
 *  data：数据。订阅后，服务端在消息可用时立即发送给客户端。事件是采用 UTF-8 编码的文本消息。事件之间由两个换行符分隔\\n\\n。每个事件由一个或多个名称：值字段组成，由单个换行符\\n 分隔。
 *  event: 事件，默认值
 *  id: 数据标识符 id 字段表示，相当于每一条数据的编号 。服务器可以发送唯一的事件标识符（id字段）。如果连接中断，客户端会自动重新连接并发送最后接收到的带有header的 Last-Event-ID 的事件 ID。
 *  retry: 重连时间 ，在重试字段中，服务器可以发送超时（以毫秒为单位），之后客户端应在连接中断时自动重新连接。如果未指定此字段，则标准应为 3000 毫秒。

#### 后端 ####

import org.springframework.http.MediaType;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.*;
    import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    
    import java.io.IOException;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * sse服务
     * 客户端关闭时，只能等待超时 服务端断开连接
     */
    @Controller
    @RequestMapping(path = "sse")
    public class SseRest {
        
    
        private final static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
    
        /**
         * 连接sse服务
         * @param id
         * @return
         * @throws IOException
         */
        @GetMapping(path = "subscribe", produces = {
        MediaType.TEXT_EVENT_STREAM_VALUE})
        public SseEmitter push(String id) throws IOException {
        
            // 超时时间设置为5分钟，用于演示客户端自动重连
            SseEmitter sseEmitter = new SseEmitter(5_60_000L);
            // 设置前端的重试时间为1s
            // send(): 发送数据，如果传入的是一个非SseEventBuilder对象，那么传递参数会被封装到 data 中
            sseEmitter.send(SseEmitter.event().reconnectTime(1000).data("连接成功"));
            sseCache.put(id, sseEmitter);
            System.out.println("add " + id);
            sseEmitter.send("你好", MediaType.APPLICATION_JSON);
            SseEmitter.SseEventBuilder data = SseEmitter.event().name("finish").id("6666").data("哈哈");
            sseEmitter.send(data);
            // onTimeout(): 超时回调触发
            sseEmitter.onTimeout(() -> {
        
                System.out.println(id + "超时");
                sseCache.remove(id);
            });
            // onCompletion(): 结束之后的回调触发
            sseEmitter.onCompletion(() -> System.out.println("完成！！！"));
            return sseEmitter;
        }
    
        /**
         * http://127.0.0.1:8080/sse/push?id=7777&content=%E4%BD%A0%E5%93%88aaaaaa
         * @param id
         * @param content
         * @return
         * @throws IOException
         */
        @ResponseBody
        @GetMapping(path = "push")
        public String push(String id, String content) throws IOException {
        
            SseEmitter sseEmitter = sseCache.get(id);
            if (sseEmitter != null) {
        
                sseEmitter.send(content);
            }
            return "over";
        }
    
        @ResponseBody
        @GetMapping(path = "over")
        public String over(String id) {
        
            SseEmitter sseEmitter = sseCache.get(id);
            if (sseEmitter != null) {
        
                // complete(): 表示执行完毕，会断开连接
                sseEmitter.complete();
                sseCache.remove(id);
            }
            return "over";
        }
    }

#### 前端 ####

有表示连接状态的 readyState 属性：  
EventSource.CONNECTING = 0 - 连接尚未建立，或已关闭且客户端正在重新连接  
EventSource.OPEN = 1 - 客户端有一个打开的连接并在接收到事件时处理它们  
EventSource.CLOSED = 2- 连接未打开，并且客户端未尝试重新连接，要么出现致命错误，要么调用了 close() 方法

要处理连接的建立，它应该订阅 onopen 事件处理程序。
    
    eventSource.onopen = function () {
        
    	console.log('connection is established');
     };

为了处理连接状态的一些异常或致命错误，它应该订阅 onerrror 事件处理程序。
    
    eventSource.onerror = function (event) {
        
    	console.log('connection state: ' + eventSource.readyState + ', error: ' + event);
     };

客户端接收消息并处理他们，可以使用onmessage方法
    
    eventSource.onmessage = function (event) {
        
    	console.log('id: ' + event.lastEventId + ', data: ' + event.data); 
    };

<!doctype html>
    <html lang="en">
    <head>
        <title>Sse测试文档</title>
    </head>
    <body>
    <div>sse测试</div>
    <div id="result"></div>
    </body>
    </html>
    <script>
        let source = null;
        let userId = 7777
        if (window.EventSource) {
          
            // 建立连接
            source = new EventSource('http://localhost:8080/sse/subscribe?id='+userId);
            console.log("连接用户=" + userId);
            /**
             * 连接一旦建立，就会触发open事件
             * 另一种写法：source.onopen = function (event) {}
             */
            source.addEventListener('open', function (e) {
          
                console.log("建立连接。。。");
            }, false);
            /**
             * 客户端收到服务器发来的数据
             * 另一种写法：source.onmessage = function (event) {}
             */
            source.addEventListener('message', function (e) {
          
                console.log(e.data);
            });
    
            source.addEventListener('finish', function (e) {
          
                console.log(e.id);
                console.log(e.data);
            });
    
        } else {
          
            console.log("你的浏览器不支持SSE");
        }
    </script>

上面的实现，用到了 SseEmitter 的几个方法，解释如下  
send(): 发送数据，如果传入的是一个非SseEventBuilder对象，那么传递参数会被封装到 data 中  
complete(): 表示执行完毕，会断开连接  
onTimeout(): 超时回调触发  
onCompletion(): 结束之后的回调触发

在实际的业务开发中，推荐使用SseEmitter，SseEmitter已经帮我们把这些封装好了

### 注意 ###

nginx 配置 proxy\_buffering off  
不配置proxy\_buffering off的话，会出现请求发出后，接口收到直接返回，无法保持长连接。  
参考网上说明：proxy\_buffering这个参数用来控制是否打开后端响应内容的缓冲区，如果这个设置为off，那么proxy\_buffers和proxy\_busy\_buffers\_size这两个指令将会失效。

问题描述  
我正在使用eventSource API，并将addEventListener()添加到eventsouce中.事件源被激活直到仅45秒.我想保持连接状态，直到服务器将响应发送回客户端.

现在，我收到以下异常消息，因为直到45秒，服务器都没有响应.

EXCEPTION: No activity within 450000 milliseconds. Reconnecting

例外:450000毫秒内没有活动.重新连接.

解决方案是定期发送数据，即使是空字节也可以正常工作，并且可以使连接保持活动状态.

如果无法建立连接，而您想重试连接，则可以使用例如设置为45秒的setTimeout.

建立连接后，请使用clearTimeout停止尝试.

[SpringBoot _SSE]: https://www.cnblogs.com/yihuihui/p/12622729.html
[SpringBoot _SSE 1]: https://blog.csdn.net/qq_43842093/article/details/125453380
[Springboot_SSE]: https://blog.csdn.net/weixin_46082019/article/details/124087719
[springboot SseEmitter]: https://blog.csdn.net/qq_39992641/article/details/105229674
[springboot_Sse_Server-sent Events_Web_-_SseEmitter]: https://blog.csdn.net/qq_50376377/article/details/126333868
[Web]: https://javaguide.cn/system-design/web-real-time-message-push.html#iframe-%E6%B5%81
[SSE_Server-Send Events]: https://zhuanlan.zhihu.com/p/444011262
[Server side event _SSE]: https://blog.csdn.net/m0_50596083/article/details/124600126
[Link 1]: https://javaguide.cn/system-design/web-real-time-message-push.html