SpringBoot SSE服务端主动推送事件详解

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

4、SSE规范

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

1）请求头
开启长连接 + 流方式传递：

Content-Type: text/event-stream;charset=UTF-8
Cache-Control: no-cache
Connection: keep-alive
1
2
3

2）数据格式
服务端发送的消息，由 message 组成，其格式如下：

field:value
1

其中 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项目，引入依赖：

        
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
            <version>2.7.12version>
        dependency>
1
2
3
4
5
6

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<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);
        };
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

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

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

2.3 前端html

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

DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Hometitle>
    <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>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

3、演示效果

参考文章：

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

– 求知若饥，虚心若愚。