SSE 推送技术

1、简介

Server-Sent Events（SSE）技术，它是一种用于实现服务器向客户端实时单向推送数据的Web技术。
SSE基于HTTP协议，允许服务器将数据以事件流（Event Stream）的形式发送给客户端。客户端通过建立持久的HTTP连接，并监听事件流，可以实时接收服务器推送的数据。
之前分享了一篇关于websocket技术的文章。本篇算是之前内容的一个补充。

官网摘要：

2、SSE和WebSocket的区别

WebSocket是另一种用于实现实时双向通信的Web技术。

数据推送方面
- SSE 是服务端像客户端的单向通信的技术。
- WebSocket是双向通讯的技术
协议方面
- SSE是基于HTTP协议的长连接，超时后可以自动重连
- WebSocket是基于ws协议的，建立双向连接实现通讯的

3、SSE的使用

SSE的使用无需引入特别的包，因为是一个Web技术，只要应为web对应的依赖即可。SSE的客户端是SseEmitter

@RestController
@RequestMapping("/foo")
public class FooController {Map<Integer,SseEmitter> map = Maps.newConcurrentMap();Map<Integer,SseEmitter> doneMap = Maps.newConcurrentMap();String curentContext = "";@GetMapping(value = "/sse", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})public SseEmitter sseEmitter(HttpServletRequest request) throws IOException {String messageId = request.getHeader("Last-Event-ID");System.out.println("Last-Event-ID 重新连接：" + messageId);Integer sseEmitterId = RandomUtils.nextInt();SseEmitter sseEmitter = new SseEmitter(15000L);System.out.println("sseEmitter 建立连接... sseEmitterId=" + sseEmitterId);map.put(sseEmitterId, sseEmitter);if (StringUtils.isNotBlank(messageId)) {if (!doneMap.containsKey(Integer.valueOf(messageId))) {sseEmitter.send(SseEmitter.event().id(messageId).data("来自客户端【" + messageId + "】补发的信息：" + curentContext));}}sseEmitter.onCompletion(() -> {System.out.println("sseEmitter 结束... sseEmitterId=" + sseEmitterId);map.remove(sseEmitterId);});return sseEmitter;}@GetMapping("/sseSend")public String sseSend() {System.out.println("获取的sseEmitter客户端：" + JSON.toJSONString(map));curentContext = "测试SSE" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss");if (!map.isEmpty()) {doneMap.clear();map.forEach((key, value) -> {try {value.send(SseEmitter.event().id(String.valueOf(key)).data("来自客户端【" + key + "】的信息：" + curentContext));doneMap.put(key, value);// value.send("来自客户端【" + key + "】的信息-----------：" + j);Thread.sleep(1000);} catch (Exception e) {throw new RuntimeException(e);}});}return "发送成功！";}@GetMapping("/closeWindow")public void test05() {System.out.println("closeWindow 窗口被关闭了....");}
}

3.1 订阅方法说明(/foo/sse)

/foo/sse 为订阅的方法。客户端打开页面，订阅该接口。返回的SseEmitter 为当前页面专属的连接，可以通过该连接推送消息。

注意事项：

订阅的返回值必须是SseEmitter ，返回的数据类型为事件流。执行返回类型的的话需要配置produces = {MediaType.TEXT_EVENT_STREAM_VALUE} 。也可以不配置，请求会自动匹配。
消息的发送，必须通过返回的SseEmitter，调用send()方法。由于需要实时推送，所以需要将创建的SseEmitter 缓存起来，随时推送消息。
SseEmitter 空参构造函数默认的超时时间为60s，也可以通过构造参数设置超时时间。案例中超时时间15s。如果设置成0，则表示永不超时。
Header中Last-Event-ID 参数为当前连接最新推送消息的ID，该ID可以自定义。消息推送之后，客户端重连之后，Header中会自动携带此参数（Last-Event-ID）。
SseEmitter 连接可以注册onCompletion【关闭】，onTimeOut【超时】,onError【错误】事件的回调。

3.2 模拟消息推送(/foo/sseSend)

/foo/sseSend 模拟消息推送的方法。获取创建的SseEmitter 连接，然后逐个推送消息。

注意事项：

map里面存放客户端的ID和连接。
因为SSE连接会超时，超时的连接关闭之后会通过回调删除连接，所以重新连接的连接不会受到消息的推送。所以使用doneMap记录已经推送的客户端。自动连接的新连接补发消息。

3.3 页面关闭事件(/foo/closeWindow)

/foo/closeWindow 是页面被关闭时的请求连接。正常的逻辑里面，应该删除服务端保存的连接。案例中没有去实现，因为页面关闭有兼容性问题。只做演示。

页面关闭的事件有兼容性问题，不能保证一定会触发
页面关闭后，推送消息的连接应该被清除，否则容易引起OOM
设置超时时间，通过回调可以避免这种问题。但是如果设置成永不超时，则会必须处理页面被关闭后连接的清除。

关闭连接也可以使用客户端close方法直接关闭，但是如果发型消息的话会报错

Caused by: java.io.IOException: 你的主机中的软件中止了一个已建立的连接。at sun.nio.ch.SocketDispatcher.write0(Native Method) ~[na:1.8.0_202]at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:51) ~[na:1.8.0_202]at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) ~[na:1.8.0_202]at sun.nio.ch.IOUtil.write(IOUtil.java:65) ~[na:1.8.0_202]at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471) ~[na:1.8.0_202]at org.apache.tomcat.util.net.NioChannel.write(NioChannel.java:135) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper.doWrite(NioEndpoint.java:1424) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.tomcat.util.net.SocketWrapperBase.doWrite(SocketWrapperBase.java:768) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.tomcat.util.net.SocketWrapperBase.flushBlocking(SocketWrapperBase.java:732) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.tomcat.util.net.SocketWrapperBase.flush(SocketWrapperBase.java:716) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.http11.Http11OutputBuffer$SocketOutputBuffer.flush(Http11OutputBuffer.java:573) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.http11.filters.ChunkedOutputFilter.flush(ChunkedOutputFilter.java:157) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.http11.Http11OutputBuffer.flush(Http11OutputBuffer.java:221) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.http11.Http11Processor.flush(Http11Processor.java:1255) [tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.AbstractProcessor.action(AbstractProcessor.java:402) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.coyote.Response.action(Response.java:209) ~[tomcat-embed-core-9.0.65.jar:9.0.65]at org.apache.catalina.connector.OutputBuffer.doFlush(OutputBuffer.java:306) ~[tomcat-embed-core-9.0.65.jar:9.0.65]... 67 common frames omitted

4、客户端的使用

不需要引入任何js，直接使用EventSource 建立连接。

案例：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>测试SSE</title>
</head>
<body><h1>测试SSE</h1><div id="stock-price"></div><div id="closeConnect">关闭连接</div>
</body>
<script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
<script>let eventSource;function init() {eventSource = new EventSource('/foo/sse');eventSource.onmessage = function (event) {console.info(event);document.getElementById('stock-price').innerHTML = event.data;};eventSource.onerror = function (event) {console.info(event.data + "::::exception");// if (event.target.readyState === EventSource.CLOSED) {//     init();// }};eventSource.addEventListener('test', e => {console.log(`message-data: ${e.data}`);}, false);}init();window.onbeforeunload  = function(e) {$.get("/foo/closeWindow", {});};$("#closeConnect").click(function(){console.info("close connection");eventSource.close();});
</script>
</html>

4.1 建立连接

new EventSource('/foo/sse') 建立连接/订阅消息，页面打开，方法执行会根据订阅的路径请求服务端获取连接。

4.2 监听消息

eventSource.onmessage 监听推送的消息，event.data直接可以获取推送的消息。
eventSource.onerror 监听异常的消息
eventSource.close() 关闭连接
eventSource.addEventListener 监听自定义时间，服务端通过SseEmitter.event().name(xxx)来设置事件的名称。