java Server Sent Event 实现消息推送

Myth丶恋晨 2024-04-22 15:13 30阅读 0赞

> 我选择的是Server-sent events)，简称SSE。主要是我理解起来简单。

这个链接是介绍 几种消息推送的方式[java实现web实时消息推送的七种方案--个人学习记录\_java实时推送前端数据\_自不惘的博客-CSDN博客][java_web_--_java_-CSDN]

## 一、java服务端代码 ##

//SSE：一种服务器发送事件(Server-sent events)
    @Slf4j
    public class SseEmitterUtil {
        //当前连接数
        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(Integer userId) {
            //设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
            SseEmitter sseEmitter =  new SseEmitter(0l);
            try{
                //注册回调：完成、失败、超时
                sseEmitter.onCompletion(completionCallBack(userId));
                sseEmitter.onError(errorCallBack(userId));
                sseEmitter.onTimeout(timeoutCallBack(userId));
                // 缓存
                sseEmitterMap.put(String.valueOf(userId), sseEmitter);
                // 数量+1
                count.getAndIncrement();
                log.info("创建新的sse连接，当前用户：{}", userId);
            }catch (Exception e){
                log.info("创建新的sse连接异常，当前用户：{}", userId);
            }
            return sseEmitter;
        }
    
        /**
         * 给指定用户发送信息
         */
        public static void sendMessage(Integer userId, String message) {
            if (!sseEmitterMap.containsKey(userId)) return;
            try {
                // sseEmitterMap.get(userId).send(message, MediaType.APPLICATION_JSON);
                sseEmitterMap.get(userId).send(message);
            } catch (IOException e) {
                log.error("用户[{}]推送异常:{}", userId, e.getMessage());
                removeUser(userId);
            }
        }
        /**
         * 移除用户连接
         */
        public static void removeUser(Integer userId) {
            sseEmitterMap.remove(userId);
            // 数量-1
            count.getAndDecrement();
            log.info("移除用户：{}", userId);
        }
        /**
         * 获取当前连接数量
         */
        public static int getUserCount() {
            return count.intValue();
        }
    
        private static Runnable completionCallBack(Integer userId) {
            return () -> {
                log.info("结束连接：{}", userId);
                removeUser(userId);
            };
        }
    
        private static Runnable timeoutCallBack(Integer userId) {
            return () -> {
                log.info("连接超时：{}", userId);
                removeUser(userId);
            };
        }
    
        private static Consumer<Throwable> errorCallBack(Integer userId) {
            return throwable -> {
                log.info("连接异常：{}", userId);
                removeUser(userId);
            };
        }
    }

## 二、前端代码 ##

目前uniapp不支持EventSource，暂时就是pc端

<!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>消息推送</title>
    </head>
    <body>
        <div>
            <button onclick="closeSse()">关闭连接</button>
            <div id="message"></div>
        </div>
    </body>
    
    <script>
        let source = null;
    
        // 用时间戳模拟登录用户
        const userId = new Date().getTime();
    
        if (window.EventSource) {
    
            // 建立连接
            source = new EventSource('http://localhost:8080/sse/connect/' + 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:8080/sse/close/' + userId, true);
            httpRequest.send();
            console.log("close");
        }
    
        // 将消息显示在网页上
        function setMessageInnerHTML(innerHTML) {
            document.getElementById('message').innerHTML += innerHTML + '<br/>';
        }
    </script>
    </html>

[java_web_--_java_-CSDN]: https://blog.csdn.net/weixin_44705301/article/details/128670957