SpringBoot SSE服务端主动推送事件详解

一、SSE概述

1、SSE简介

SSE(Server Sent Event)，直译为服务器发送事件，也就是服务器主动发送事件，客户端可以获取到服务器发送的事件。

我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕。但是在SSE的使用场景下，客户端发起请求，然后建立SEE连接一直保持，服务端就可以返回数据给客户端。

SSE简单来说就是服务器主动向前端推送数据的一种技术，它是单向的。SSE适用于消息推送，监控等只需要服务器推送数据的场景中。比如：文件下载时，后端可以推送下载进度条信息。

2、特点

SSE (Server Send Event)服务端主动推送：

html5新标准，用来从服务端实时推送数据到浏览器端，
直接建立在当前http连接上，本质上是保持一个http长连接，轻量协议

简单的服务器数据推送的场景，使用服务器推送事件还是很方便的。

3、SSE和WebScoket的区别

SSE 是单通道，只能服务端向客户端发消息；而WebScoket 是双通道。
SSEWebScokethttp 协议独立的 websocket 协议轻量，使用简单相对复杂默认支持断线重连需要自己实现断线重连文本传输二进制传输支持自定义发送的消息类型-

4、SSE规范

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

1）请求头
开启长连接 + 流方式传递：

Content-Type: text/event-stream;charset=UTF-8Cache-Control: no-cache
Connection: keep-alive

2）数据格式
服务端发送的消息，由 message 组成，其格式如下：

field:value

其中 field 有五种可能：

空：即以:开头，表示注释，可以理解为服务端向客户端发送的心跳，确保连接不中断
data：数据
event：事件，默认值
id：数据标识符用 id 字段表示，相当于每一条数据的编号
retry：重连时间

二、SSE实战

使用 SpringBoot简单实现一个SSE服务端主动推送数据为前端，前端页面接受后展示进度条。

1、SseEmitter类简介

SpringBoot 利用 SseEmitter 来支持SSE，并对SSE规范做了一些封装，使用起来非常简单。

我们操作SseEmitter对象，关注消息文本即可。

SseEmitter类的几个方法：

send()：发送数据，如果传入的是一个非SseEventBuilder对象，那么传递参数会被封装到 data 中。
complete()：表示执行完毕，会断开连接。
onTimeout()：连接超时时回调触发。
onCompletion()：结束之后的回调触发。
onError()：报错时的回调触发。

2、示例实战

创建 SpringBoot项目，引入依赖：

<!-- 里面包含了 spring-webmvc--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><version>2.7.12</version></dependency>

2.1 创建 SseServer

我们创建一个 SseServer来简单封装一下业务操作SSE的方法。

importorg.springframework.web.servlet.mvc.method.annotation.SseEmitter;/**
 * SseServer业务封装类来操作SEE
 */@Slf4jpublicclassSseServer{/**
     * 当前连接总数
     */privatestaticAtomicInteger currentConnectTotal =newAtomicInteger(0);/**
     * messageId的 SseEmitter对象映射集
     */privatestaticMap<String,SseEmitter> sseEmitterMap =newConcurrentHashMap<>();/**
     * 创建sse连接
     *
     * @param messageId - 消息id（唯一）
     * @return
     */publicstaticSseEmittercreateConnect(String messageId){/**
         * 设置连接超时时间。0表示不过期，默认是30秒，超过时间未完成会抛出异常
         */SseEmitter sseEmitter =newSseEmitter(0L);/*
        // 超时时间设置为3s，设置前端的重试时间为1s。重连时，注意总数的统计
        SseEmitter sseEmitter = new SseEmitter(3_000L);
        try {
            sseEmitter.send(
                    SseEmitter.event()
                    .reconnectTime(1000L)
                    //.data("前端重连成功") // 重连成功的提示信息
            );
        } catch (IOException e) {
            log.error("前端重连异常 ==> messageId={}, 异常信息：", messageId, e.getMessage());
            e.printStackTrace();
        }*/// 注册回调
        sseEmitter.onCompletion(completionCallBack(messageId));
        sseEmitter.onTimeout(timeOutCallBack(messageId));
        sseEmitter.onError(errorCallBack(messageId));
        sseEmitterMap.put(messageId, sseEmitter);//记录一下连接总数。数量+1int count = currentConnectTotal.incrementAndGet();
        log.info("创建sse连接成功 ==> 当前连接总数={}， messageId={}", count, messageId);return sseEmitter;}/**
     * 给指定 messageId发消息
     *
     * @param messageId - 消息id（唯一）
     * @param message   - 消息文本
     */publicstaticvoidsendMessage(String messageId,String message){if(sseEmitterMap.containsKey(messageId)){try{
                sseEmitterMap.get(messageId).send(message);}catch(IOException e){
                log.error("发送消息异常 ==> messageId={}, 异常信息：", messageId, e.getMessage());
                e.printStackTrace();}}else{thrownewRuntimeException("连接不存在或者超时， messageId="+ messageId);}}/**
     * 给所有 messageId广播发送消息
     *
     * @param message
     */publicstaticvoidbatchAllSendMessage(String message){
        sseEmitterMap.forEach((messageId, sseEmitter)->{try{
                sseEmitter.send(message,MediaType.APPLICATION_JSON);}catch(IOException e){
                log.error("广播发送消息异常 ==> messageId={}, 异常信息：", messageId, e.getMessage());removeMessageId(messageId);}});}/**
     * 给指定 messageId集合群发消息
     *
     * @param messageIds
     * @param message
     */publicstaticvoidbatchSendMessage(List<String> messageIds,String message){if(CollectionUtils.isEmpty(messageIds)){return;}// 去重
        messageIds = messageIds.stream().distinct().collect(Collectors.toList());
        messageIds.forEach(userId ->sendMessage(userId, message));}/**
     * 给指定组群发消息（即组播，我们让 messageId满足我们的组命名确定即可）
     *
     * @param groupId
     * @param message
     */publicstaticvoidgroupSendMessage(String groupId,String message){if(MapUtils.isEmpty(sseEmitterMap)){return;}
        sseEmitterMap.forEach((messageId, sseEmitter)->{try{// 这里 groupId作为前缀if(messageId.startsWith(groupId)){
                    sseEmitter.send(message,MediaType.APPLICATION_JSON);}}catch(IOException e){
                log.error("组播发送消息异常 ==> groupId={}, 异常信息：", groupId, e.getMessage());removeMessageId(messageId);}});}/**
     * 移除 MessageId
     *
     * @param messageId
     */publicstaticvoidremoveMessageId(String messageId){
        sseEmitterMap.remove(messageId);//数量-1
        currentConnectTotal.getAndDecrement();
        log.info("remove messageId={}", messageId);}/**
     * 获取所有的 MessageId集合
     *
     * @return
     */publicstaticList<String>getMessageIds(){returnnewArrayList<>(sseEmitterMap.keySet());}/**
     * 获取当前连接总数
     *
     * @return
     */publicstaticintgetConnectTotal(){return currentConnectTotal.intValue();}/**
     * 断开SSE连接时的回调
     *
     * @param messageId
     * @return
     */privatestaticRunnablecompletionCallBack(String messageId){return()->{
            log.info("结束连接 ==> messageId={}", messageId);removeMessageId(messageId);};}/**
     * 连接超时时回调触发
     *
     * @param messageId
     * @return
     */privatestaticRunnabletimeOutCallBack(String messageId){return()->{
            log.info("连接超时 ==> messageId={}", messageId);removeMessageId(messageId);};}/**
     * 连接报错时回调触发。
     *
     * @param messageId
     * @return
     */privatestaticConsumer<Throwable>errorCallBack(String messageId){return throwable ->{
            log.error("连接异常 ==> messageId={}", messageId);removeMessageId(messageId);};}}

2.2 业务controller

@RestController@CrossOrigin@RequestMapping("/sse")publicclassSseDemoController{/**
     * 用户SSE连接
     * 它返回一个SseEmitter实例，这时候连接就已经创建了.
     *
     * @return
     */@GetMapping("/userConnect")publicSseEmitterconnect(){/**
         * 一般取登录用户账号作为 messageId。分组的话需要约定 messageId的格式。
         * 这里模拟创建一个用户连接
         */String userId ="userId-"+RandomUtils.nextInt(1,10);returnSseServer.createConnect(userId);}/**
     * 模拟实例：下载进度条显示。 前端访问下载接口之前，先建立用户SSE连接，然后访问下载接口，服务端推送消息。
     * http://localhost:8080/sse/downLoad/userId-1
     *
     * @throws InterruptedException
     */@GetMapping("/downLoad/{userId}")publicvoidpushOne(@PathVariable("userId")String userId)throwsInterruptedException{for(int i =0; i <=100; i++){if(i >50&& i <70){Thread.sleep(500L);}else{Thread.sleep(100L);}System.out.println("sendMessage --> 消息="+ i);SseServer.sendMessage(userId,String.valueOf(i));}System.out.println("下载成功");}/**
     * 广播发送。http://localhost:8080/sse/pushAllUser
     *
     * @throws InterruptedException
     */@GetMapping("/pushAllUser")publicvoidpushAllUser()throwsInterruptedException{for(int i =0; i <=100; i++){if(i >50&& i <70){Thread.sleep(500L);}else{Thread.sleep(100L);}System.out.println("batchAllSendMessage --> 消息="+ i);SseServer.batchAllSendMessage(String.valueOf(i));}}}

2.3 前端html

简单写一个html来演示效果。

<!DOCTYPEhtml><htmllang="en"><head><metacharset="UTF-8"><title>Home</title><script>var sseSource =newEventSource("http://localhost:8080/sse/userConnect");<!-- 添加一个信息回调 -->
        sseSource.onmessage=function(event){
            console.log("test=>",event)
            document.getElementById("result").innerText = event.data+'%';
            document.getElementById("my-progress").value = event.data;}// 使用vue交互事件，可以添加一些SSE的回调// sseSource.dispatchEvent();// sseSource.close();</script></head><body><divid="result"></div>
        下载进度条：<progressstyle="width: 300px"id="my-progress"value="0"max="100"></progress></body></html>

3、演示效果

在这里插入图片描述

参考文章：

HTML- server-sent-events：https://html.spec.whatwg.org/multipage/server-sent-events.html

– 求知若饥，虚心若愚。

标签： SpringBoot SSE SSE服务端主动推送事件详解

本文转载自: https://blog.csdn.net/qq_42402854/article/details/130932936
版权归原作者 Charge8 所有，如有侵权，请联系我们删除。

SpringBoot SSE服务端主动推送事件详解

一、SSE概述

1、SSE简介

2、特点

3、SSE和WebScoket的区别

4、SSE规范

二、SSE实战

1、SseEmitter类简介

2、示例实战

2.1 创建 SseServer

2.2 业务controller

2.3 前端html

3、演示效果

发表评论

“SpringBoot SSE服务端主动推送事件详解”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航