SSE（ServerSendEvents）实践

　　介绍
　　HTTP 是客户端-服务器计算模型中的请求-响应协议。要开始交换，客户端向服务器提交请求。为了完成交换，服务器向客户端返回响应。服务器只能向一个客户端发送响应 （发出请求的那个） 。在 HTTP 协议中，客户端是消息交换的发起者。
　　有些场景需要由服务端主动推送消息给客户端。实现这一点的方法之一是允许服务器在 发布/订阅 计算模型中向客户端推送消息。要开始交换，客户端从服务器订阅消息。在交换期间，服务器向许多订阅的客户端发送消息（一旦它们可用）。
　　服务器发送事件 (SSE) 是一种简单的技术，用于为特定的 Web 应用程序实现服务器到客户端的异步通信。 概述
　　有多种技术允许客户端从服务器接收有关异步更新的消息。它们可以分为两类： 客户端拉取 和 服务器推送 。 客户端拉取
　　在客户端拉取技术中，客户端会定期向服务器请求更新。服务器可以使用更新或尚未更新的特殊响应进行响应。有两种类型的客户端拉取：短轮询和长轮询。 短轮询
　　客户端定期向服务器发送请求。如果服务器有更新，它会向客户端发送响应并关闭连接。如果服务器没有更新，它也会向客户端发送一个响应并关闭连接。 长轮询
　　客户端向服务器发送请求。如果服务器有更新，它会向客户端发送响应并关闭连接。如果服务器没有更新，它会保持连接直到更新可用。当更新可用时，服务器向客户端发送响应并关闭连接。如果更新在某个超时时间内不可用，服务器会向客户端发送响应并关闭连接。 服务端推送
　　在服务器推送技术中，服务器在消息可用后立即主动向客户端发送消息。其中，有两种类型的服务器推送：SSE和 WebSocket。 SSE（Server-Send Events）
　　SSE 是一种在基于浏览器的 Web 应用程序中仅从服务器向客户端发送文本消息的技术。SSE基于 HTTP 协议中的持久连接， 具有由 W3C 标准化的网络协议和 EventSource 客户端接口，作为 HTML5 标准套件的一部分。 WebSocket
　　WebSocket 是一种在 Web 应用程序中实现同时、双向、实时通信的技术。WebSocket 基于 HTTP 以外的协议（TCP），因此可能需要额外设置网络基础设施（代理服务器、NAT、防火墙等）。
　　客户端通过Http协议请求，在握手阶段升级为WebSocket协议。 SSE 网络协议
　　要订阅服务器事件，客户端发出 GET 请求带有指定的header： Accept: text/event-stream 表示可接收事件流类型 Cache-Control: no-cache 禁用任何的事件缓存 Connection: keep-alive 表示正在使用持久连接 GET /sse HTTP/1.1
　　Accept: text/event-stream
　　Cache-Control: no-cache
　　Connection: keep-alive
　　服务器应该使用带有标题的响应来确认订阅： Content-Type: text/event-stream;charset=UTF-8  表示标准要求的事件的媒体类型和编码 Transfer-Encoding: chunked  表示服务器流式传输动态生成的内容，因此内容大小事先未知 HTTP/1.1 200
　　Content-Type: text/event-stream;charset=UTF-8
　　Transfer-Encoding: chunked
　　订阅后，服务端在消息可用时立即发送给客户端。事件是采用 UTF-8 编码的文本消息。事件之间由两个换行符分隔  。每个事件由一个或多个名称：值字段组成，由单个换行符  分隔。
　　在数据字段中，服务器可以发送事件数据 data: The first event.
　　data: The second event.
　　服务器可以发送唯一的事件标识符（id字段）。如果连接中断，客户端会 自动重新连接 并发送最后接收到的带有header的 Last-Event-ID 的事件 ID。
　　在事件字段中，服务器可以发送事件类型。服务器可以在同一个订阅中发送不同类型的事件，也可以不发送任何类型的事件。 event: type1
　　data: An event of type1.
　　event: type2
　　data: An event of type2.
　　data: An event without any type.
　　在重试字段中，服务器可以发送超时（以毫秒为单位），之后客户端应在连接中断时自动重新连接。如果未指定此字段，则标准应为 3000 毫秒。 retry: 1000
　　如果一行以冒号字符 : 开头，客户端应该忽略它。这可用于从服务器发送评论或防止某些代理服务器因超时关闭连接。
　　: pingSSE 客户端: EventSource 接口
　　要打开连接，应创建一个 EventSource 对象。 var eventSource = new EventSource(＂/sse);
　　尽管 SSE 旨在将事件从服务器发送到客户端，但可以使用 GET 查询参数将数据从客户端传递到服务器。 var eventSource = new EventSource(＂/sse?event=type1);
　　...
　　eventSource.close();
　　eventSource = new EventSource(＂/sse?event=type1&event=type2);
　　...
　　要关闭连接，应调用方法 close()。
　　eventSource.close();
　　有表示连接状态的 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);
　　};
　　SSE可被大多数浏览器支持：
　　SSE Java 服务端: Spring Web MVC介绍
　　Spring Web MVC 框架 5.2.0 是基于 Servlet 3.1 API 且用线程池实现异步应用程序. 所以应用能够被使用在 Servlet 3.1+ 的容器，比如：Tomcat 8.5 和 Jetty 9.3. 概述
　　使用Spring MVC来发送事件: 使用 @RestController 注解创建一个控制器类（Controller） 创建一个方法来创建一个客户端连接，它返回一个  SseEmitter ，处理 GET 请求并产生（ produces ）文本/事件流  (text/event-stream) 创建一个新的 SseEmitter, 保存它并从方法中返回 在另一个线程中异步发送事件, 先拿到保存的 SseEmitter  并根据需要多次调用 调用SseEmitter.send 方法 完成事件发送, 调用  SseEmitter.complete() 方法 要异常完成发送事件，请调用 SseEmitter.completeWithError() 方法
　　示例： @RestController public class SseWebMvcController      private SseEmitter emitter;      @GetMapping(path=＂/sse＂, produces=MediaType.TEXT_EVENT_STREAM_VALUE)     SseEmitter createConnection() {         emitter = new SseEmitter();         return emitter;     }      // in another thread     void sendEvents() {         try {             emitter.send(＂Alpha＂);             emitter.send(＂Omega＂);              emitter.complete();         } catch(Exception e) {             emitter.completeWithError(e);         }     } }
　　处理短暂的周期性事件流
　　在这个例子中，服务器每秒发送一个持续时间短的周期性事件流 - 一个有限的词流，直到词完成。
　　示例： @Controller @RequestMapping(＂/sse/mvc＂) public class WordsController {     private static final String[] WORDS = ＂The quick brown fox jumps over the lazy dog.＂.split(＂ ＂);     private final ExecutorService cachedThreadPool = Executors.newCachedThreadPool();     @GetMapping(path = ＂/words＂, produces = MediaType.TEXT_EVENT_STREAM_VALUE)    SseEmitter getWords() {        SseEmitter emitter = new SseEmitter();         cachedThreadPool.execute(() -> {            try {                for (int i = 0; i < WORDS.length; i++) {                    emitter.send(WORDS[i]);                    TimeUnit.SECONDS.sleep(1);                }                 emitter.complete();            } catch (Exception e) {                emitter.completeWithError(e);            }        });         return emitter;    } }
　　运行效果：
　　客户端示例（words.html）：               
　　运行效果：
　　处理长期持续的周期性事件
　　在此示例中，服务器发送持久的周期性事件流 - 每秒可能无限的服务器性能信息流： import org.springframework.http.MediaType; 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 javax.annotation.PostConstruct; import java.io.IOException; import java.util.UUID; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit;  @RestController @RequestMapping(＂/sse/mvc＂) public class LongEventController {      private final ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);      private SseEmitter emitter;      @PostConstruct     public void init() {         scheduledThreadPool.scheduleAtFixedRate(() -> {             try {                 if (emitter != null) {                     emitter.send(UUID.randomUUID().toString());                 }             } catch (IOException e) {                 e.printStackTrace();             }         }, 0, 1, TimeUnit.SECONDS);     }      @GetMapping(path = ＂/getEvents＂, produces = MediaType.TEXT_EVENT_STREAM_VALUE)     public SseEmitter getEvents() {         emitter = new SseEmitter();         return emitter;     } }
　　效果预览（每秒输出一次）：
　　处理非周期性事件
　　非周期性是指没有固定的时间周期，可能由其他因素在任意时刻都可能触发，下面示例通过spring event来模拟触发因子。@RestController @RequestMapping(＂/sse/mvc＂) public class EventController {       private SseEmitter emitter;     @Autowired     private ApplicationContext applicationContext;      /**      *  订阅事件通道      * @return      */     @GetMapping(path = ＂/event＂, produces = MediaType.TEXT_EVENT_STREAM_VALUE)     public SseEmitter event() {         emitter = new SseEmitter();         return emitter;     }      /**      *  模拟某一事件触发动作      * @param eventType      */     @GetMapping(path = ＂/trigger＂)     public void trigger(String eventType) {         applicationContext.publishEvent(new MyEvent(eventType));     }      /**      *  监听动作，发送给客户端数据      */     @EventListener(classes = MyEvent.class)     public void triggerEvent(MyEvent event) throws IOException {         emitter.send(event);     } }
　　效果:
　　模拟触发动作：调用 http://localhost:8080/sse/mvc/trigger?eventType=customer
　　客户端收到数据：
　　SSE Java 服务端: Spring Web Flux介绍
　　Spring Web Flux 框架 5.2.0 是基于 Reactive Streams API 且使用  event-loop  计算模型来实现异步java应用程序。 此类应用程序可以在非阻塞 Web 服务器（例如 Netty 4.1 和 Undertow 1.4）和 Servlet 3.1+ 容器（例如 Tomcat 8.5 和 Jetty 9.3）上运行。 概述
　　使用 Spring Web Flux 框架实现发送事件： 使用 @RestController 注解创建一个控制器类（Controller） 创建一个方法来创建一个客户端连接，它返回一个  Flux ，处理 GET 请求并产生（ produces ）文本/事件流  (text/event-stream) 创建一个新的 Flux对象且在方法中返回它
　　简单示例： @RestController public class ExampleController      @GetMapping(path=＂/sse＂, produces=MediaType.TEXT_EVENT_STREAM_VALUE)     public Flux createConnectionAndSendEvents() {         return Flux.just(＂Alpha＂, ＂Omega＂);     } }
　　处理短暂的周期性事件流
　　和上面spring mvc的示例一样，也是每秒输出数据，实现如下： @GetMapping(path = ＂/words＂, produces = MediaType.TEXT_EVENT_STREAM_VALUE)     public Flux getWords() {         return Flux                 .zip(Flux.just(WORDS), Flux.interval(Duration.ofSeconds(1)))                 .map(Tuple2::getT1);     }Flux.interval(Duration.ofSeconds(1)) 表示 每秒钟发出一次递增的 long 值 通过 zip 方法将它们组合在一起以输入 map(Tuple2::getT1) 表示 提取元组的第一个元素
　　效果：
　　处理长期持续的周期性事件
　　对比spring mvc的实现，我们改为flux实现，如下： @GetMapping(path = ＂/getEvents＂, produces = MediaType.TEXT_EVENT_STREAM_VALUE)     public Flux getEvents() {         return Flux                 .interval(Duration.ofSeconds(1))                 .map(sequence -> UUID.randomUUID().toString());     }
　　效果和上面是一样的，可以看出，reactive api是非常的简洁。
　　SSE的缺点或限制它是单向的，只能由服务端发送给客户端 只发送文本消息；尽管可以使用 Base64 编码和 gzip 压缩来发送二进制消息，但效率很低。 许多浏览器允许打开数量非常有限的 SSE 连接（Chrome、Firefox 每个浏览器最多 6 个连接）