一、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 是双通道。

  SSE	WebScoket
  http 协议	独立的 websocket 协议
  轻量，使用简单	相对复杂
  默认支持断线重连	需要自己实现断线重连
  文本传输	二进制传输
  支持自定义发送的消息类型	-

  4、SSE规范

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

  1）请求头

  开启长连接 + 流方式传递：

  Content-Type: text/event-stream;charset=UTF-8
  Cache-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项目，引入依赖：

              
              
                  org.springframework.boot
                  spring-boot-starter-web
                  2.7.12
              
      

      2.1 创建 SseServer

      我们创建一个 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 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 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 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 errorCallBack(String messageId) {
              return throwable -> {
                  log.error("连接异常 ==> messageId={}", messageId);
                  removeMessageId(messageId);
              };
          }
      }
      

      2.2 业务controller

      @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));
              }
          }
      }
      

      2.3 前端html

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

      
      
      
          
          
          
      
      
          
      
              下载进度条：
          
      
      

      3、演示效果

      

      参考文章：

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

        – 求知若饥，虚心若愚。