Server side event (SSE)实现消息推送功能

问题场景

 在开发web项目时，有一个需求是：后端服务器要主动地、不断地推送消息给客户端网页。要实现该需求，需要先考虑几个常用的技术方案：

• 在客户端网页用fetch、XmlHttpRequest发送请求是行不通的，因为这类请求在后端返回一次数据之后就会中断连接，导致后端无法主动地传数据给客户端。
• 客户端网页使用轮询或者长轮询的资源效率低，会花很多时间在不断地建立网络连接、不断地断开网络连接。（不熟悉轮询和长轮询的朋友可以自己百度一下，有很多博文）。
• 客户端和后端一同使用WebSocket的话，虽然只需要建立一次网络连接就能让客户端和后端都主动地、不断地向对方发送消息，但该项目只需要后端向客户端主动地、不断地发送消息，且WebSocket使用起来相对复杂，因此没必要用它。
• 客户端和后端一同使用Server side event (SSE)的话，能在建立一次网络连接后，让后端向客户端主动地、不断地发送消息，并且使用起来也非常简单，是实现项目功能的首选方案。

Server side event (SSE)简介

 SSE定义了客户端网页和后端服务器的网络连接方式，具体包括：

1. 先由客户端发送一次http请求给后端（注意，第一次http请求的发起者永远是客户端而不是后端）。
2. 后端接收请求并建立网络连接，该网络连接将长久存在，可以一直被使用，直到下述第4点发生。
3. 在网络连接后，后端不会立马发送数据给客户端，而是在后端运行到了自定义的条件时（如数据库里更新了数据），再将数据通过网络连接发送给客户端。接着后端会继续等待这个自定义的条件的发生。因此，后端能够主动地、不断地向客户端发送数据。
4. 有4种让网络连接断开的情况：
   ①网络问题（如客户端断网），浏览器监测到网络问题后，每隔一段时间就会向后端发起重连接请求；后端是察觉不到客户端已经断网了的，只有在后端向客户端发送数据，但收到了发送错误的信息后，才会在后端也断开连接。
   ②网络连接的时长超出了后端设置的超时时间，此时后端会断开网络连接，并且通知客户端也要断开，但这之后客户端仍然会不断地发起重连接请求，以便继续传输数据。
   ③后端主动地调用SSE的断开函数，此时客户端和后端的网络连接就会断开，客户端仍然会不断重连。
   ④客户端主动地调用SSE的断开函数，此时客户端和后端的网络连接就会断开，但是客户端不会再重连。
   一旦客户端重连成功，情况又变到了上面的第2点的位置。

在后端使用SSE

 个人使用的后端框架是Springboot，该框架自带实现了SSE的类，我们只需要如下4步即可让后端能实现SSE功能。

1. 在项目的pom.xml文件中导入Springboot的web依赖，从而导入实现了SSE功能的类。

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-web</artifactId>
</dependency>

2. 在项目的Controller类里定义返回SseEmitter对象的函数，来让Springboot自动建立客户端和后端的SSE连接。

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
SseEmitter emitter; // 声明SseEmitter类型的变量
@GetMapping("/test/sse")
@ResponseBody // @ResponseBody让函数返回的对象不被解析成前端文件的路径
public SseEmitter handle(){
    emitter=new SseEmitter();
    return emitter;
}

只要客户端访问了/test/sse这个路径，Springboot会根据返回的SseEmitter对象，来自动建立起客户端和后端的SSE连接，不需要我们再来做建立网络连接的事。

3. 在项目的Controller类里定义让SseEmitter发送数据的函数。

public void send() throws IOException{
      emitter.send(
          SseEmitter
          .event()
          .data("<Your Data>")
          .id("<Your Id>")
          .reconnectTime(<Your Time>)
      );

其中SseEmitter.event().data("<Your Data>").id("<Your Id>").reconnectTime()得到的类型是SseEventBuilder类，用来定义要发送的数据。然后调用SseEmitter的send(SseEventBuilder)方法，SseEmitter便会自动地把定义好的数据发送给客户端。说一句，SseEmitter发送数据不一定要靠send(SseEventBuilder)方法，只是个人觉得这样比较方便且功能完备，至于SseEmitter还有哪些发送数据的方法，直接看官方的SseEmitter文档吧。

 接下来说一下SSE发送的数据包含哪些东西。就拿上述SseEmitter.event().data("<Your Data>").id("<Your Id>").reconnectTime()定义的数据来说：

1. SseEmitter.event()用来得到一个记录数据的容器（该容器使用建造者模式添加数据），该方法不带任何参数。
2. .data("<Your Data>")用"<Your Data>"来添加传输给客户端的数据，参数是Object类型，但最好以字符串为参数，这样就不需要担心SseEmitter会怎样处理Object了（如无需担心SseEmitter怎么自动处理Map）。如果是想返回JSON数据，可先将JSON数据转换为字符串；如果是想返回二进制数据，可以用base64的编码方式，每6个比特用一个字符来代替，再由客户端将字符解码为比特。
3. .id("<Your Id>")用"<Your Id>"来作为这条数据的id。每当客户端要重连时（由于断网、或者网络连接时长达到了后端设置的超时时间等），客户端就会将最后收到的那条数据的id，连同重连接请求一并发给后端，从而让后端知道客户端已经接收了哪些数据。后端可以在HttpServletRequest的请求头中拿到这个id，至于拿到id之后怎么处理，就需要后端自定义函数了。举两个例子，一个例子是后端拿到id后不做任何处理的，另一个例子是后端拿到id后，对比自己已产生的id，来推断出客户端漏接收了哪些数据，然后将客户端漏收的数据补发出去。

// 例子1：后端拿到id后不做任何处理
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

SseEmitter emitter; // 声明SseEmitter类型的变量
@GetMapping("/test/sse")
@ResponseBody
// 客户端每次重连，都是在触发该函数，因此在该函数用HttpServletRequest即可获取请求里的id
public SseEmitter handle(HttpServletRequest request){
	String lastId=request.getHeader("Last-Event-ID");
	// 虽然获得了该id，但不做任何事情，客户端有没有漏收数据就不管了 
    emitter=new SseEmitter();
    return emitter;
}

// 例子2：后端拿到id后，看客户端漏收了哪些数据，漏收的要补发

SseEmitter emitter; // 声明SseEmitter类型的变量
Integer backEndId; // 记录后端最新数据的id
@GetMapping("/test/sse")
@ResponseBody
// 客户端每次重连，都是在触发该函数，因此在该函数用HttpServletRequest即可获取请求里的id
public SseEmitter handle(HttpServletRequest request){
	String lastId=request.getHeader("Last-Event-ID");
	// 获得了客户端最后收到的数据的id后，对比后端最新数据的id，来补发客户端漏收的数据
	// 自定义一个函数来决定要补发的数据，如定义下面的compare()函数
	String data=compare(lastId,backEndId) 
    emitter=new SseEmitter();
    // 先添加要发送的数据给SseEmitter ，等Springboot自动建立SSE连接后便会自动发送该数据
    emitter.send(
          SseEmitter
          .event()
          .data(data)
          .id("<Your Id>")
          .reconnectTime(5000)
      );
    return emitter;
}

• .reconnectTime(<Your Time>)用<Your Time>定义在网络连接断开后，客户端向后端发起重连的时间间隔（以毫秒为单位），网络连接断开并重连的情况见开头那几点。这个东西的作用就是让客户端在网络连接断开的情况下重连后端，恢复数据传输。

  既然说到了客户端的时间设置.reconnectTime(<Your Time>)，那么另外一个和时间相关的设置就是后端SseEmitter对象的超时时间。在用new SseEmitter()实例化SseEmitter 对象时，还可以输入一个long型参数（如new SseEmitter(timeout); ），代表后端SseEmitter 对象的超时时间（以毫秒为单位），不传入参数的话则默认是30000毫秒。如果SseEmitter 对象的存在时间超过了设定的超时时间，那么后端和客户端分别发生以下事情：
  

  • 后端的SseEmitter 对象会主动通知客户端已超时，并且这个SseEmitter 对象已无法再发送新数据（此时用send()方法发送会抛异常）。
  • 客户端方面，如果后端给客户端发送过数据，那么客户端接收到超时通知后会一直自动重连（也就是重新访问"/test/sse"路径），重连的时间间隔为.reconnectTime(<Your Time>)中设定的时间，在得到一个新的SseEmitter对象后继续发送数据。但如果后端从未用SseEmitter对象发送过任何数据，那么客户端便不会自动重连，而是直接报503的http错误代码。
  • 说一下超时时间的作用，后端设置超时时间，可以在客户端一直断网、直接关闭页面但未提醒后端的情况下，让后端在一定时间等待后自动关闭网络连接，节省资源。如果不存在超时时间的话，一旦遇到以上意外情况，后端就会一直维持该网络连接，直到某一次发送数据发现连接不上才会去断开后端的连接，这会浪费资源。

客户端使用SSE

 客户端和后端建立SSE连接，需要双方都做出努力，不是某一方做了另一方就不用做。基于Springboot的后端如何建立SSE连接已经在上面讲了，这部分就讲下客户端的Javascript该如何建立SSE连接。
 客户端需要做3件事：

1. 实例化支持SSE功能的对象，在绝大部分浏览器都内置了EventSource对象，可以用于实现SSE功能。（好用的Chrome, Firefox, Edge当然内置了该对象；但IE浏览器没内置该对象，不过这对IE浏览器来说还挺合理）

// new一个EventSource对象，第一个参数是后端的访问地址；第二个参数是可选的，如果要填就只能填{withCredentials:true}或{withCredentials:true}，表示发送或不发送Cookie
var eventSource = new EventSource("http://127.0.0.1:8000/test/sse");

这一行就可以让客户端向后端发送建立SSE连接的请求，只要后端成功建立了SSE连接，后端就可以主动向客户端主动、不断地发送数据了。

2. 监听 EventSource对象接收数据、报错的事件。

// 监听接受数据的事件
eventSource.addEventListener("message", function(event) {console.log(event.data)});

EventSource默认支持三类事件：“open”, “message”, “error”；分别表示客户端和后端建立了连接、客户端接收到了来自后端的数据、客户端报错这三个场景。使用addEventListener()用于注册事件，第一个参数是事件名，第二个参数是一个回调函数，这个回调函数自带一个MessageEvent 对象，MessageEvent.data 表示后端传来的数据，MessageEvent.lastEventId能表示后端传来的数据id。客户端拿到MessageEvent.data 后，就可以用来更新用户的页面了，不一定非要像console.log(event.data)在控制台打印。

3. 在任务结束后关闭SSE连接。下面这行会让客户端自己关闭SSE连接，并且也会通知后端关闭SSE连接。

eventSource.close()

实例

为了展示客户端和后端的SSE通信，实现这样一个简单的例子：

1. 客户端用EventSource发起创建SSE连接的请求给后端，后端进行接收并建立起SSE连接。

2. 后端每隔5秒向客户端发送一次数据，并设置id用于展示如何补发数据，设置reconnectTime来展示控制客户端重连接的时间间隔（间隔太低浪费资源，太高又可能浪费时间来等待）。

3. 后端SseEmitter 的对象设置超时时间为30秒，用来展示客户端在SSE连接超时后如何反应。（“超时时间的大小怎么设置”要看后端传数据一般需要的时长，太短了就会让客户端多次重连接后端消耗资源，太长了又可能在客户端断网、关闭页面但未通知后端的情况下，依然让后端维持太久的连接。）

代码1：pom.xml中的依赖

后端只需要引入springboot的web场景。

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-web</artifactId>
</dependency>

代码2：后端的controller

先说一句题外话，前后端分离项目先解决跨域问题，自己用浏览器做测试也要解决跨域问题。对于Springboot框架来说，解法跨域的方案之一就是继承WebMvcConfigurationSupport 类并注册到容器中（加上@Configuration或@Component等注册组件），并重写和跨域相关的方法。这样的话Springboot会自动读取这个组件对跨域的设置。代码如下

@Configuration
public class MvcConfig extends WebMvcConfigurationSupport  {
	// 重写这个方法
    @Override
    protected void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/**")
                .allowedOrigins("http://localhost:8080") // 允许本地8080端口进行请求
                .allowedMethods("POST", "GET", "PUT", "OPTIONS", "DELETE")
                .allowCredentials(true)
                .allowedHeaders("*")
                .maxAge(3600); 
    }
}

让后端能建立SSE连接，并每隔5秒向客户端发送数据。

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
public class TestSSEController {
    SseEmitter emitter; // 实现SSE功能的类
    
    Integer backEndId=0; // 记录后端最后发送的数据的id，用于判断客户端是否遗漏数据
    String backEndData=null; // 记录后端最后发送的数据，如果客户端有遗漏数据，则先发送该条数据

    @GetMapping("/test/sse")
    @ResponseBody
    public SseEmitter handle(HttpServletRequest request) throws IOException{
        emitter=new SseEmitter(30000); // 设置后端SSE连接的超时时长为30秒
        String lastId=request.getHeader("Last-Event-ID"); // 获取客户端收到的最后一个数据的id，用来判断是否有漏收
        // 两种情况意味着漏收数据：
        // ①客户端没收到任何数据，但后端记录的最后发送的数据不为空
        // ②客户端收到过数据，但客户端最后收到的数据id和后端最后发送的数据id不一致
        if((lastId==null && backEndData!=null) || (lastId!=null && !lastId.equals(backEndId.toString()))){
        	// 客户端漏收了数据，后端先将最新的数据返回给客户端
            emitter.send(
                SseEmitter
                .event()
                .data(backEndData)
                .id(backEndId.toString())
                .reconnectTime(3000)
            );
        }
        return emitter;
    }

	// Springboot用@Scheduled创建定时任务，每隔5秒自动执行向客户端发送数据的函数
	// 除了要在这里加@Scheduled注解，还要在启动类（通常是xxxApplication类）上加@EnableScheduling注解
    @Scheduled(cron="0/5 * *  * * ? ")
    public void trigger() throws IOException{
        try{
            backEndId++; // 每次发送时id都要+1
            emitter.send(
                SseEmitter
                .event()
                .data("data"+backEndId) // 发送新数据
                .id(backEndId.toString()) // 发送新id
                .reconnectTime(3000)
            );
            backEndData="data"+backEndId; // 更新最新的数据
            System.out.println("id: "+backEndId.toString()+" data: "+backEndData);
        }catch(Exception e){
            backEndId--; 
            System.out.println("id: "+backEndId.toString()+" data: "+backEndData);
            System.out.println(e.getMessage());
        }
    }
}

代码3：客户端的设置

首先要做的不是打开浏览器在控制台输入javascript，而是开启一个客户端服务器，以便和后端通信。最简单的开启方式就是用python自带的功能，可以创建一个位于localhost:8080的客户端服务器（要为该路径设置跨域，怎么设置的见上面）。

python -m http.server 8080
# 这个8080的端口号可以改成别的

开启客户端服务器后，打开浏览器访问localhost:8080路径，再打开控制台，用EventSource对象发起创建SSE连接的请求，并且在每次收到新数据时打印数据。

var eventSource = new EventSource("http://localhost:8000/test/sse");
eventSource.addEventListener("message", function(e) {console.log(e.data)});
// 这两行一执行，客户端就会向后端发起创建SSE的请求，后端接收后会成功建立SSE。
// 之后后端每隔5秒发送数据，客户端就会把这些数据打印在控制台中

效果1：控制台打印的数据的样子

后端每隔5秒发送一次数据给客户端，客户端拿到之后打印到控制台，所以也是差不多每隔5秒打印一次数据。

效果2：实际发起了多次SSE连接

我们将后端SSE超时时间设置为了30秒，且每5秒发送一次数据，那么后端大约会发6个数据，那为什么客户端控制台上打印了远远不止这么多的数据呢？这时候就需要看网络请求而不是控制台了，在开发者工具栏里点击“网络”，结果如下两图所示。可以看到每一个SSE请求只会带来6个数据，我们之所以能收到超出6个数据，是因为客户端自动地重连了，让后端不断地产生新的SseEmitter（忘记这回事的朋友看上文）。

效果3：客户端断网重连时补发数据

为了模拟客户端断网重连，后端补发遗漏数据的情况。我在传送到第15个数据时启动了浏览器的离线功能来模拟断网，见下图红色圈圈那里。

此时客户端检测到了网络问题，会每隔3秒（3秒是后端的.reconnectTime(30000)设置的）重连一次，此时控制台会显示以下重连信息：

后端只有在发送数据后，收到数据没传过去的错误信息时才会意识到客户端那边已断开连接，此时后端也会断开自己的连接，SseEmitter也不能再传数据（此时再调用SseEmitter.send()方法会抛异常的），而在此之前后端可能已经发了好几个新数据，如下图所示。客户端在收到第15个数据的时候就已经断网了，但后端不能立即知道这件事，就又发到了第17个数据，这时才收到数据传输的错误信息，从而断开连接，并让SseEmitter再也不能发送任何消息。

接下来就是展示补发数据的过程了，如果恢复网络，也就是将下图红圈圈的下拉框选成“高速3G”，客户端重连后端就会成功，上文后端Controller实现了补发最后一个数据的功能，此时客户端会收到第17个数据，正如下图红圈圈所示。如果你想要补发中途漏掉的所有数据，那就自己修改后端的Controller。

说几句废话，上图网络请求中未成功的、标红的，都是客户端重连失败的产物；至于为什么上图只有3个数据而不是预想中的6个，是因为我提前调用了客户端中EventSource对象的close()方法，来关闭SSE连接。

线程安全问题

上面说的都是测试性的例子，没有讨论后端SseEmitter在多线程并发下的问题，这里补充一下：
SseEmitter对象的send()方法是线程不安全的，如果两个线程对同一个SseEmitter对象交替使用send()方法，就可能会让发送的数据有问题。所以，如果想实现线程安全，要么一个线程修改一个SseEmitter对象而不跟其他线程共用，要么自己写一个类继承SseEmitter，重写send()方法时加上synchronized同步代码块，如下面所示

// 自己封装一个线程安全的SseEmitter
public class SSEThreadSafeWrapper extends SseEmitter{
    public SSEThreadSafeWrapper() {
        super();
    }
    public SSEThreadSafeWrapper(long l) {
        super(l);
    }
    @Override
    public void send(SseEventBuilder arg0) throws IOException {
    // 线程安全
        synchronized(this){
            super.send(arg0);
        }
    }
    @Override
    public void send(Object object) throws IOException {
    // 线程安全
        synchronized(this){
            super.send(object);
        }
    }
    @Override
    public void send(Object object, MediaType mediaType) throws IOException {
    // 线程安全
        synchronized(this){
            super.send(object, mediaType);
        }
    }
}

然后Springboot就改一处地方：文章来源地址https://www.toymoban.com/news/detail-456357.html

emitter=new SseEmitter(30000);
// 上面的改为下面的
emitter=new SSEThreadSafeWrapper (30000); 

到了这里，关于Server side event (SSE)实现消息推送功能的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！