SpringBoot使用SSE推送消息

2020-11-11

字数统计: 2.1k字 | 阅读时长≈ 9分

序言

服务器推送技术在一些需要实时数据的开发场景会被广泛使用，要获得实时数据，前端轮询的方式明显是被摒弃的，大部分人第一时间想到的就是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秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        sseEmitterMap.put(userId, sseEmitter);
        // 数量+1
        count.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);
        // 数量-1
        count.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>