赞
踩
SSE(Server Sent Event),直译为服务器发送事件,也就是服务器主动发送事件,客户端可以获取到服务器发送的事件。
我们常见的 http 交互方式是客户端发起请求,服务端响应,然后一次请求完毕。但是在SSE的使用场景下,客户端发起请求,然后建立SEE连接一直保持,服务端就可以返回数据给客户端。
SSE简单来说就是服务器主动向前端推送数据的一种技术,它是单向的。SSE适用于消息推送,监控等只需要服务器推送数据的场景中。比如:文件下载时,后端可以推送下载进度条信息。
SSE (Server Send Event)服务端主动推送:
简单的服务器数据推送的场景,使用服务器推送事件还是很方便的。
SSE 是单通道,只能服务端向客户端发消息;而WebScoket 是双通道。
SSE | WebScoket |
---|---|
http 协议 | 独立的 websocket 协议 |
轻量,使用简单 | 相对复杂 |
默认支持断线重连 | 需要自己实现断线重连 |
文本传输 | 二进制传输 |
支持自定义发送的消息类型 | - |
在 html5 的定义中,服务端SSE,一般需要遵循以下要求:
1)请求头
开启长连接 + 流方式传递:
Content-Type: text/event-stream;charset=UTF-8
Cache-Control: no-cache
Connection: keep-alive
2)数据格式
服务端发送的消息,由 message 组成,其格式如下:
field:value
其中 field 有五种可能:
使用 SpringBoot简单实现一个SSE服务端主动推送数据为前端,前端页面接受后展示进度条。
SpringBoot 利用 SseEmitter 来支持SSE,并对SSE规范做了一些封装,使用起来非常简单。
我们操作SseEmitter对象,关注消息文本即可。
SseEmitter类的几个方法:
创建 SpringBoot项目,引入依赖:
<!-- 里面包含了 spring-webmvc-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.7.12</version>
</dependency>
我们创建一个 SseServer来简单封装一下业务操作SSE的方法。
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; /** * SseServer业务封装类来操作SEE */ @Slf4j public class SseServer { /** * 当前连接总数 */ private static AtomicInteger currentConnectTotal = new AtomicInteger(0); /** * messageId的 SseEmitter对象映射集 */ private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>(); /** * 创建sse连接 * * @param messageId - 消息id(唯一) * @return */ public static SseEmitter createConnect(String messageId) { /** * 设置连接超时时间。0表示不过期,默认是30秒,超过时间未完成会抛出异常 */ SseEmitter sseEmitter = new SseEmitter(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); //记录一下连接总数。数量+1 int count = currentConnectTotal.incrementAndGet(); log.info("创建sse连接成功 ==> 当前连接总数={}, messageId={}", count, messageId); return sseEmitter; } /** * 给指定 messageId发消息 * * @param messageId - 消息id(唯一) * @param message - 消息文本 */ public static void sendMessage(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 { throw new RuntimeException("连接不存在或者超时, messageId=" + messageId); } } /** * 给所有 messageId广播发送消息 * * @param message */ public static void batchAllSendMessage(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 */ public static void batchSendMessage(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 */ public static void groupSendMessage(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 */ public static void removeMessageId(String messageId) { sseEmitterMap.remove(messageId); //数量-1 currentConnectTotal.getAndDecrement(); log.info("remove messageId={}", messageId); } /** * 获取所有的 MessageId集合 * * @return */ public static List<String> getMessageIds() { return new ArrayList<>(sseEmitterMap.keySet()); } /** * 获取当前连接总数 * * @return */ public static int getConnectTotal() { return currentConnectTotal.intValue(); } /** * 断开SSE连接时的回调 * * @param messageId * @return */ private static Runnable completionCallBack(String messageId) { return () -> { log.info("结束连接 ==> messageId={}", messageId); removeMessageId(messageId); }; } /** * 连接超时时回调触发 * * @param messageId * @return */ private static Runnable timeOutCallBack(String messageId) { return () -> { log.info("连接超时 ==> messageId={}", messageId); removeMessageId(messageId); }; } /** * 连接报错时回调触发。 * * @param messageId * @return */ private static Consumer<Throwable> errorCallBack(String messageId) { return throwable -> { log.error("连接异常 ==> messageId={}", messageId); removeMessageId(messageId); }; } }
@RestController @CrossOrigin @RequestMapping("/sse") public class SseDemoController { /** * 用户SSE连接 * 它返回一个SseEmitter实例,这时候连接就已经创建了. * * @return */ @GetMapping("/userConnect") public SseEmitter connect() { /** * 一般取登录用户账号作为 messageId。分组的话需要约定 messageId的格式。 * 这里模拟创建一个用户连接 */ String userId = "userId-" + RandomUtils.nextInt(1, 10); return SseServer.createConnect(userId); } /** * 模拟实例:下载进度条显示。 前端访问下载接口之前,先建立用户SSE连接,然后访问下载接口,服务端推送消息。 * http://localhost:8080/sse/downLoad/userId-1 * * @throws InterruptedException */ @GetMapping("/downLoad/{userId}") public void pushOne(@PathVariable("userId") String userId) throws InterruptedException { 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") public void pushAllUser() throws InterruptedException { 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)); } } }
简单写一个html来演示效果。
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Home</title> <script> var sseSource = new EventSource("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> <div id="result"></div> 下载进度条:<progress style="width: 300px" id="my-progress" value="0" max="100"></progress> </body> </html>
参考文章:
– 求知若饥,虚心若愚。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。