介绍

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 编码的文本消息。事件之间由两个换行符分隔\n\n。每个事件由一个或多个名称：值字段组成，由单个换行符\n 分隔。

在数据字段中，服务器可以发送事件数据

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

如果一行以冒号字符 : 开头，客户端应该忽略它。这可用于从服务器发送评论或防止某些代理服务器因超时关闭连接。

: ping

SSE 客户端: 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来发送事件:

1. 使用 @RestController 注解创建一个控制器类（Controller）
2. 创建一个方法来创建一个客户端连接，它返回一个 SseEmitter，处理 GET 请求并产生（produces）文本/事件流 (text/event-stream)

3. 创建一个新的 SseEmitter, 保存它并从方法中返回
4. 在另一个线程中异步发送事件, 先拿到保存的 SseEmitter 并根据需要多次调用调用SseEmitter.send 方法

5. 完成事件发送, 调用 SseEmitter.complete() 方法
6. 要异常完成发送事件，请调用 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）：

<!DOCTYPE html>
<html lang="en">
<head>
   <meta charset="UTF-8">
   <title>Server-Sent Events client example with EventSource</title>
</head>
<body>
<script>
   if (window.EventSource == null) {
       alert('The browser does not support Server-Sent Events');
   } else {
       var eventSource = new EventSource('/sse/mvc/words');

       eventSource.onopen = function () {
           console.log('connection is established');
       };

       eventSource.onerror = function (error) {
           console.log('connection state: ' + eventSource.readyState + ', error: ' + event);
       };

       eventSource.onmessage = function (event) {
           console.log('id: ' + event.lastEventId + ', data: ' + event.data);

           if (event.data.endsWith('.')) {
               eventSource.close();
               console.log('connection is closed');
           }
       };
   }
</script>
</body>
</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 框架实现发送事件：

1. 使用 @RestController 注解创建一个控制器类（Controller）
2. 创建一个方法来创建一个客户端连接，它返回一个 Flux，处理 GET 请求并产生（produces）文本/事件流 (text/event-stream)

3. 创建一个新的 Flux对象且在方法中返回它

简单示例：

@RestController
public class ExampleController

    @GetMapping(path="/sse", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> createConnectionAndSendEvents() {
        return Flux.just("Alpha", "Omega");
    }
}

处理短暂的周期性事件流

和上面spring mvc的示例一样，也是每秒输出数据，实现如下：

 @GetMapping(path = "/words", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> 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<String> getEvents() {
        return Flux
                .interval(Duration.ofSeconds(1))
                .map(sequence -> UUID.randomUUID().toString());
    }

效果和上面是一样的，可以看出，reactive api是非常的简洁。

SSE的缺点或限制

• 它是单向的，只能由服务端发送给客户端
• 只发送文本消息；尽管可以使用 Base64 编码和 gzip 压缩来发送二进制消息，但效率很低。

• 许多浏览器允许打开数量非常有限的 SSE 连接（Chrome、Firefox 每个浏览器最多 6 个连接）