SpringBoot使用SSE推送消息

序言

服务器推送技术在一些需要实时数据的开发场景会被广泛使用，要获得实时数据，前端轮询的方式明显是被摒弃的，大部分人第一时间想到的就是websocket。之前的一个项目也是用的websocket，但是后期造成了会话缓存不断变大的问题，花了很多时间来解决。因此我就想着找一个替代websocket，简单好用的技术，因此便发现了SSE。以下记录学习SSE踩的坑，尤其看了很多博客和教学视频，带来的SSE就是一直轮询的疑惑。

如果对websocket有兴趣，参考博客：SpringBoot2.0集成WebSocket，实现后台向前端推送信息

SSE：Server send Event：服务端发送事件【划重点，服务端推送，而不是轮询请求】。

1、错误用法

之所以叫错误用法，是我觉得这不是SSE的真正使用方式，而不是使用错误。

当你打算了解SSE技术，百度搜了各种博客或者视频，他们会给你一个Controller代码，告诉你SSE的消息格式是 "data:" + 消息 + "\n\n"，然后写个页面来接收消息，再贴个F12控制台的图片说SSE就是轮询请求，比如下面的：
Controller代码：

@GetMapping(value = "/simpleSse", produces = "text/event-stream; charset=utf-8")
@ResponseBody
public String simpleSse() {return "data: xxx-xxx" + Math.random() + "\n\n";
}

页面：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>Title</title>
</head>
<body><div>sse测试</div><div id="result"></div>
</body>
<script>var source = new EventSource('http://localhost:7788/sse/simpleSse');source.onmessage = function (event) {text = document.getElementById('result').innerText;text += '\n' + event.data;document.getElementById('result').innerText = text;};<!-- 添加一个开启回调 -->source.onopen = function (event) {text = document.getElementById('result').innerText;text += '\n 开启: ';console.log(event);document.getElementById('result').innerText = text;};
</script>
</html>

页面结果：

这个结果就是我根据网上很多博客和视频实现的效果，乍一看，是页面不停的请求才实现了数据刷新的效果。但是，不是说好了SSE是一种服务端向页面推送数据的技术吗？这明明就是页面在发请求啊？？还好我之前用过websocket，不然还真是被忽悠了，破绽就在页面数据打印上面一行的“开启”两个字，这是在open（也就是创建页面与服务端连接）的时候才会有的，按理说只会在进入页面就与服务器创建连接，仅一次（如果没有设置超时时间的话），而不是一直在连接。
之前使用websocket的时候，大概流程应该是：打开页面，页面会以一个id(或者称之为通道)与服务端连接，而服务端就可以通过这个id来向页面推送信息。而以上事例的代码完全没有体现出服务端推送消息的点在哪里？？如果SSE仅限于此，那它还有存在的必要吗？还不如直接在页面上进行定时请求。

2、正确使用

上面的事例明显不能满足我对SSE的期望，继续百度，终于让我找到了我想要的真正的SSE。
参考博客：springboot SseEmitter 消息推送
这里我们不再使用原生的SSE，而是经过封装SseEmitter（就在web包下，不需要添加新的依赖），使用SseEmitter我们不再需要在@GetMapping上写produces = “text/event-stream; charset=utf-8”，也不需要在返回数据里拼"data:“和”\n\n"的格式。
工具类:

public class SseEmitterServer {private static final Logger logger = LoggerFactory.getLogger(SseEmitterServer.class);/*** 当前连接数*/private static AtomicInteger count = new AtomicInteger(0);/*** 使用map对象，便于根据userId来获取对应的SseEmitter，或者放redis里面*/private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();/*** 创建用户连接并返回 SseEmitter** @param userId 用户ID* @return SseEmitter*/public static SseEmitter connect(String userId) {// 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutExceptionSseEmitter sseEmitter = new SseEmitter(0L);// 注册回调sseEmitter.onCompletion(completionCallBack(userId));sseEmitter.onError(errorCallBack(userId));sseEmitter.onTimeout(timeoutCallBack(userId));sseEmitterMap.put(userId, sseEmitter);// 数量+1count.getAndIncrement();logger.info("创建新的sse连接，当前用户：{}", userId);return sseEmitter;}/*** 给指定用户发送信息  -- 单播*/public static void sendMessage(String userId, String message) {if (sseEmitterMap.containsKey(userId)) {try {// sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);System.out.println(userId + "==" + message);sseEmitterMap.get(userId).send(message);} catch (IOException e) {logger.error("用户[{}]推送异常:{}", userId, e.getMessage());removeUser(userId);}}}/*** 向多人发布消息   -- 组播* @param groupId 开头标识* @param message 消息内容*/public static void groupSendMessage(String groupId, String message) {if (MapUtil.isNotEmpty(sseEmitterMap)) {/*Set<String> ids = sseEmitterMap.keySet().stream().filter(m -> m.startsWith(groupId)).collect(Collectors.toSet());batchSendMessage(message, ids);*/sseEmitterMap.forEach((k, v) -> {try {if (k.startsWith(groupId)) {v.send(message, MediaType.APPLICATION_JSON);}} catch (IOException e) {logger.error("用户[{}]推送异常:{}", k, e.getMessage());removeUser(k);}});}}/*** 群发所有人   -- 广播*/public static void batchSendMessage(String message) {sseEmitterMap.forEach((k, v) -> {try {v.send(message, MediaType.APPLICATION_JSON);} catch (IOException e) {logger.error("用户[{}]推送异常:{}", k, e.getMessage());removeUser(k);}});}/*** 群发消息*/public static void batchSendMessage(String message, Set<String> ids) {ids.forEach(userId -> sendMessage(userId, message));}/*** 移除用户连接*/public static void removeUser(String userId) {sseEmitterMap.remove(userId);// 数量-1count.getAndDecrement();logger.info("移除用户：{}", userId);}/*** 获取当前连接信息*/public static List<String> getIds() {return new ArrayList<>(sseEmitterMap.keySet());}/*** 获取当前连接数量*/public static int getUserCount() {return count.intValue();}private static Runnable completionCallBack(String userId) {return () -> {logger.info("结束连接：{}", userId);removeUser(userId);};}private static Runnable timeoutCallBack(String userId) {return () -> {logger.info("连接超时：{}", userId);removeUser(userId);};}private static Consumer<Throwable> errorCallBack(String userId) {return throwable -> {logger.info("连接异常：{}", userId);removeUser(userId);};}
}

有部分博客在创建连接的时候，设置的超时时间太短，造成页面一直重连，就会出现和上面错误用法类似的情况，页面一直在请求接口。

页面：

<!DOCTYPE html>
<html lang="en"><head><meta charset="UTF-8"><title>SseEmitter</title>
</head><body>
<button onclick="closeSse()">关闭连接</button>
<div id="message"></div>
</body>
<script>let source = null;// 用时间戳模拟登录用户const userId = new Date().getTime();if (window.EventSource) {// 建立连接source = new EventSource('http://localhost:7788/sse/connect/' + userId);setMessageInnerHTML("连接用户=" + userId);/*** 连接一旦建立，就会触发open事件* 另一种写法：source.onopen = function (event) {}*/source.addEventListener('open', function(e) {setMessageInnerHTML("建立连接。。。");}, false);/*** 客户端收到服务器发来的数据* 另一种写法：source.onmessage = function (event) {}*/source.addEventListener('message', function(e) {setMessageInnerHTML(e.data);});/*** 如果发生通信错误（比如连接中断），就会触发error事件* 或者：* 另一种写法：source.onerror = function (event) {}*/source.addEventListener('error', function(e) {if (e.readyState === EventSource.CLOSED) {setMessageInnerHTML("连接关闭");} else {console.log(e);}}, false);} else {setMessageInnerHTML("你的浏览器不支持SSE");}// 监听窗口关闭事件，主动去关闭sse连接，如果服务端设置永不过期，浏览器关闭后手动清理服务端数据window.onbeforeunload = function() {closeSse();};// 关闭Sse连接function closeSse() {source.close();const httpRequest = new XMLHttpRequest();httpRequest.open('GET', 'http://localhost:7788/sse/close/' + userId, true);httpRequest.send();console.log("close");}// 将消息显示在网页上function setMessageInnerHTML(innerHTML) {document.getElementById('message').innerHTML += innerHTML + '<br/>';}
</script></html>

项目启动成功之后，我使用了四个浏览器进行访问，分别是谷歌、360急速、星愿和微软的edge，遗憾的是edge不支持SSE。

测试单个发送：

测试多个发送

测试发送所有

3、总结

相对于可以双向（服务端和客户端）的websocket，SSE仅能实现服务端向客户端推送消息，但是一般情况也足够使用了，使用方面也是比较简单的，以上内容仅是我个人在学习SSE中的看法。