消息推送之SSE

🕗 发布于 2024-11-13 23:42 java SSE

一、简介

市面上很多系统都有
在这里插入图片描述
小铃铛

以上三种的消息提醒。但大多分为2类，一类移动端，一类web端比，通常在服务端会有若干张消息推送表，用来记录用户触发不同事件所推送不同类型的消息，前端主动查询（拉）或者被动接收（推）用户所有未读的消息数。消息推送无非是推（push）和拉（pull）两种形式
然而较常见的几种方案设计分别为：

1.短轮询

轮询(polling)应该是实现消息推送方案中最简单的一种，这里我们暂且将轮询分为短轮询和长轮询。
短轮询很好理解，指定的时间间隔，由浏览器向服务器发出HTTP请求，服务器实时返回未读消息数据给客户端，浏览器再做渲染显示。
一个简单的JS定时器就可以搞定，每秒钟请求一次未读消息数接口，返回的数据展示即可。
在这里插入图片描述

问题：效果还是可以的，短轮询实现固然简单，缺点也是显而易见，由于推送数据并不会频繁变更，无论后端此时是否有新的消息产生，客户端都会进行请求，势必会对服务端造成很大压力，浪费带宽和服务器资源。

2.长轮询

长轮询是对上边短轮询的一种改进版本，在尽可能减少对服务器资源浪费的同时，保证消息的相对实时性。长轮询在中间件中应用的很广泛，比如Nacos和apollo配置中心，消息队列kafka、RocketMQ中都有用到长轮询。
这次我使用apollo配置中心实现长轮询的方式，应用了一个类DeferredResult，它是在servelet3.0后经过Spring封装提供的一种异步请求机制，直意就是延迟结果。
在这里插入图片描述
DeferredResult可以允许容器线程快速释放占用的资源，不阻塞请求线程，以此接受更多的请求提升系统的吞吐量，然后启动异步工作线程处理真正的业务逻辑，处理完成调用DeferredResult.setResult(200)提交响应结果。

下边我们用长轮询来实现消息推送。

因为一个ID可能会被多个长轮询请求监听，所以我采用了guava包提供的Multimap结构存放长轮询，一个key可以对应多个value。一旦监听到key发生变化，对应的所有长轮询都会响应。前端得到非请求超时的状态码，知晓数据变更，主动查询未读消息数接口，更新页面数据。

@Controller
@RequestMapping("/polling")
public class PollingController {

    // 存放监听某个Id的长轮询集合
    // 线程同步结构
    public static Multimap<String, DeferredResult<String>> watchRequests = Multimaps.synchronizedMultimap(HashMultimap.create());

    /**
     * 设置监听
     */
    @GetMapping(path = "watch/{id}")
    @ResponseBody
    public DeferredResult<String> watch(@PathVariable String id) {
        // 延迟对象设置超时时间
        DeferredResult<String> deferredResult = new DeferredResult<>(TIME_OUT);
        // 异步请求完成时移除 key，防止内存溢出
        deferredResult.onCompletion(() -> {
            watchRequests.remove(id, deferredResult);
        });
        // 注册长轮询请求
        watchRequests.put(id, deferredResult);
        return deferredResult;
    }

    /**
     * 变更数据
     */
    @GetMapping(path = "publish/{id}")
    @ResponseBody
    public String publish(@PathVariable String id) {
        // 数据变更 取出监听ID的所有长轮询请求，并一一响应处理
        if (watchRequests.containsKey(id)) {
            Collection<DeferredResult<String>> deferredResults = watchRequests.get(id);
            for (DeferredResult<String> deferredResult : deferredResults) {
                deferredResult.setResult("我更新了" + new Date());
            }
        }
        return "success";
    }

当请求超过设置的超时时间，会抛出AsyncRequestTimeoutException异常，这里直接用@ControllerAdvice全局捕获统一返回即可，前端获取约定好的状态码后再次发起长轮询请求，如此往复调用。

@ControllerAdvice
public class AsyncRequestTimeoutHandler {

    @ResponseStatus(HttpStatus.NOT_MODIFIED)
    @ResponseBody
    @ExceptionHandler(AsyncRequestTimeoutException.class)
    public String asyncRequestTimeoutHandler(AsyncRequestTimeoutException e) {
        System.out.println("异步请求超时");
        return "304";
    }
}

我们来测试一下，首先页面发起长轮询请求/polling/watch/10086监听消息更变，请求被挂起，不变更数据直至超时，再次发起了长轮询请求；紧接着手动变更数据/polling/publish/10086，长轮询得到响应，前端处理业务逻辑完成后再次发起请求，如此循环往复。

长轮询相比于短轮询在性能上提升了很多，但依然会产生较多的请求，这是它的一点不完美的地方。

3.MQTT

什么是 MQTT协议？
MQTT 全称(Message Queue Telemetry Transport)：一种基于发布/订阅（publish/subscribe）模式的轻量级通讯协议，通过订阅相应的主题来获取消息，是物联网（Internet of Thing）中的一个标准传输协议。
该协议将消息的发布者（publisher）与订阅者（subscriber）进行分离，因此可以在不可靠的网络环境中，为远程连接的设备提供可靠的消息服务，使用方式与传统的MQ有点类似。
在这里插入图片描述

TCP协议位于传输层，MQTT 协议位于应用层，MQTT 协议构建于TCP/IP协议上，也就是说只要支持TCP/IP协议栈的地方，都可以使用MQTT协议。
为什么要用 MQTT协议？
MQTT协议为什么在物联网（IOT）中如此受偏爱？而不是其它协议，比如我们更为熟悉的 HTTP协议呢？
首先HTTP协议它是一种同步协议，客户端请求后需要等待服务器的响应。而在物联网（IOT）环境中，设备会很受制于环境的影响，比如带宽低、网络延迟高、网络通信不稳定等，显然异步消息协议更为适合IOT应用程序。
HTTP是单向的，如果要获取消息客户端必须发起连接，而在物联网（IOT）应用程序中，设备或传感器往往都是客户端，这意味着它们无法被动地接收来自网络的命令。
通常需要将一条命令或者消息，发送到网络上的所有设备上。HTTP要实现这样的功能不但很困难，而且成本极高。
具体的MQTT协议介绍和实践，这里我就不再赘述了，大家可以参考我之前的两篇文章，里边写的也都很详细了。

MQTT协议的介绍
我也没想到 springboot + rabbitmq 做智能家居，会这么简单
MQTT实现消息推送
未读消息（小红点），前端与 RabbitMQ 实时消息推送实践，贼简单~

4.Websocket

websocket应该是大家都比较熟悉的一种实现消息推送的方式，上边我们在讲SSE的时候也和websocket进行过比较。
WebSocket是一种在TCP连接上进行全双工通信的协议，建立客户端和服务器之间的通信渠道。浏览器和服务器仅需一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。
在这里插入图片描述

springboot整合websocket，先引入websocket相关的工具包

<!-- 引入websocket -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

服务端使用@ServerEndpoint注解标注当前类为一个websocket服务器，客户端可以通过ws://localhost:7777/webSocket/10086来连接到WebSocket服务器端。

@Component
@Slf4j
@ServerEndpoint("/websocket/{userId}")
public class WebSocketServer {
    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    private static final CopyOnWriteArraySet<WebSocketServer> webSockets = new CopyOnWriteArraySet<>();
    // 用来存在线连接数
    private static final Map<String, Session> sessionPool = new HashMap<String, Session>();
    /**
     * 链接成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId) {
        try {
            this.session = session;
            webSockets.add(this);
            sessionPool.put(userId, session);
            log.info("websocket消息: 有新的连接，总数为:" + webSockets.size());
        } catch (Exception e) {
        }
    }
    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("websocket消息: 收到客户端消息:" + message);
    }
    /**
     * 此为单点消息
     */
    public void sendOneMessage(String userId, String message) {
        Session session = sessionPool.get(userId);
        if (session != null && session.isOpen()) {
            try {
                log.info("websocket消: 单点消息:" + message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

前端初始化打开WebSocket连接，并监听连接状态，接收服务端数据或向服务端发送数据。

<script>
    var ws = new WebSocket('ws://localhost:7777/webSocket/10086');
    // 获取连接状态
    console.log('ws连接状态：' + ws.readyState);
    //监听是否连接成功
    ws.onopen = function () {
        console.log('ws连接状态：' + ws.readyState);
        //连接成功则发送一个数据
        ws.send('test1');
    }
    // 接听服务器发回的信息并处理展示
    ws.onmessage = function (data) {
        console.log('接收到来自服务器的消息：');
        console.log(data);
        //完成通信后关闭WebSocket连接
        ws.close();
    }
    // 监听连接关闭事件
    ws.onclose = function () {
        // 监听整个过程中websocket的状态
        console.log('ws连接状态：' + ws.readyState);
    }
    // 监听并处理error事件
    ws.onerror = function (error) {
        console.log(error);
    }
    function sendMessage() {
        var content = $("#message").val();
        $.ajax({
            url: '/socket/publish?userId=10086&message=' + content,
            type: 'GET',
            data: { "id": "7777", "content": content },
            success: function (data) {
                console.log(data)
            }
        })
    }
</script>

页面初始化建立websocket连接，之后就可以进行双向通信了，效果还不错
在这里插入图片描述
然而如果只需要实现上面简单的站内信的消息推送，我这边尝试的是SSE

5.站内信推荐型SSE

服务端向客户端推送消息，其实除了可以用WebSocket这种耳熟能详的机制外，还有一种服务器发送事件(Server-sent events)，简称SSE。
SSE它是基于HTTP协议的，我们知道一般意义上的HTTP协议是无法做到服务端主动向客户端推送消息的，但SSE是个例外，它变换了一种思路。

在这里插入图片描述

SSE在服务器和客户端之间打开一个单向通道，服务端响应的不再是一次性的数据包而是text/event-stream类型的数据流信息，在有数据变更时从服务器流式传输到客户端。
整体的实现思路有点类似于在线视频播放，视频流会连续不断的推送到浏览器，你也可以理解成，客户端在完成一次用时很长（网络不畅）的下载。

具体实现：

Spring中有一个SseEmitter类。这个类就是一个长连接的文件流，它是由父类【ResponseBodyEmitter】继成而来，从SseEmitter中可以看到send()的方法借用了SseEventBuilder接口实现发送，SseEventBuilder中的data默认采用的是Json形式，当然，咱们也可以自己设置一种格式进行数据推送。然后就是SseEmitter的静态块，采用的是：
TEXT_PLAIN = new MediaType(“text”, “plain”, StandardCharsets.UTF_8);
这说明 SseEmitter 使用Text的方式进行通讯推送。

编写一个客户端，模拟请求：

<!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>
    // 用时间戳模拟登录用户
    const userId = new Date().getTime();
    // 替换为你的token
    const token = 'x5hVfCwzdaFkxQQrGlDVSR9jGHBNacKO';
    // 使用Fetch API来发送请求，并监听响应
    // const url = 'http://localhost:9101/sse/subscribe?id=1679048475536338945';
    // const url = '/application/api/webapp/message/info/sse?userId=1679048475536338945';
    const url = 'http://localhost:8080/sse?userId=1679048475536338945';

    if (!!window.EventSource) {

        // 创建一个EventSource对象
        const source = new EventSource(url);

        /**
         * 连接一旦建立，就会触发open事件
         * 另一种写法：source.onopen = function (event) {}
         */
        source.addEventListener('open', function (e) {
            setMessageInnerHTML("建立连接。。。");
        }, false);

        /**
         * 客户端收到服务器发来的数据
         * 另一种写法：source.onmessage = function (event) {}
         */
        source.addEventListener('message', function (e) {
            setMessageInnerHTML(e.data);
        });
          
          //主动断开链接
        // source.close()
        
        /**
         * 如果发生通信错误（比如连接中断），就会触发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/CloseSseConnect/?clientId=e410d4c1d71c469b8d719de5d39783b7', true);
        httpRequest.send();
        console.log("close");
    }

    // 将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }


    /*
     * 如果前端采用vue 等框架，需要传递token则可以使用如下代码
     * 1.引入EventSourcePolyfill包 ：npm install event-source-polyfill --save
     * 2.引入包 import { EventSourcePolyfill } from "event-source-polyfill";
     * 3.官方文档：https://www.npmjs.com/package/event-source-polyfill
     */
    // const eventSource = new EventSourcePolyfill('/application/api/webapp/message/info/sse?userId=1679048475536338945', {
    //     headers: {
    //         'token': tool.data.get('TOKEN'),
    //         'Content-Type': 'text/event-stream',
    //         'charset': 'UTF-8',
    //         'Cache-Control':' no-cache',
    //         'Connection': 'keep-alive'
    //     }
    // })
    //
    // eventSource.addEventListener('open', () => {
    //     console.log('连接成功')
    // });
    //
    // eventSource.addEventListener('message', (event) => {
    //     console.log('消息:', event.data)
    // })
    //
    // eventSource.addEventListener('error', (event) => {
    //     if (event.readyState === EventSource.CLOSED) {
    //         console.error('关闭')
    //     } else {
    //         console.error('连接错误', event)
    //     }
    // })

</script>
</html>

编写服务端

package com.example.messagedemo.demo;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.example.messagedemo.demos.web.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * sse推送消息
 * @author jiangjian
 */
@Slf4j
@RestController
public class DevSseMessageController {
    Map<String, SseEmitter> emitterMap = new ConcurrentHashMap<>();

    /**
     * 建立连接
     */
    @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @CrossOrigin(value = "*")
    public ResponseBodyEmitter sseServer(@RequestParam String userId) throws IOException {
        SseEmitter emitter = null;
        if ((emitter = emitterMap.get(userId)) != null){
            emitter.send("重连");
            return emitter;
        }
        // 判断其他服务器有没有对应的连接，有的话，就算了。直接返回。或者直接转发。可以通过直接调用或者通过mq推送之类的
        emitter = new SseEmitter(300000L);
        // 当连接超时时触发
        emitter.onTimeout(()->{
            emitterMap.remove(userId);
            log.info("timeout");
        });
        // 在连接完成时候触发，可在连接完成时执行一些清理工作
        emitter.onCompletion(()->{
            emitterMap.remove(userId);
            log.info("completion");
        });
        // 在客户端断开连接的时候会触发error回调，当连接异常时触发
        emitter.onError(e->{
            emitterMap.remove(userId);
            log.error("error",e);
        });
        log.info("create for {}",userId);
        emitterMap.put(userId, emitter);
        return emitter;
    }
    /**
     * 推送消息，只需要通过emitter发送即可
     */
    @GetMapping("/send")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        send(message);
        return ResponseEntity.ok("ok");
    }


    /**
     * 定期清理不活跃客户端
     * 容器一加载久会定期清理不活跃的连接，避免内存溢出
     */
    @PostConstruct
    private void scheduleCleanup() {
        new ScheduledThreadPoolExecutor(1).scheduleAtFixedRate(()->{
            emitterMap.values().removeIf(emitter -> {
                try{
                   // ping(emitter);
                    ping2(emitter);
                    return false;
                }catch (IOException e){
                    log.warn("清理一个不活跃的客户端");
                    return true;
                }
            });
        }, 0, 10, TimeUnit.SECONDS);
    }

   //定期查询
    private void ping(SseEmitter emitter) throws IOException {
        Set<ResponseBodyEmitter.DataWithMediaType> dataWithMediaTypes = SseEmitter.event()
                .id(UUID.randomUUID().toString())
                .name("ping")
                //.data("ping")
                .data("哈哈哈哈哈哈哈哈哈哈哈啊哈哈哈哈哈哈哈")
                .comment("comment")
                .build();
        emitter.send(dataWithMediaTypes);
    }

    private void ping2(SseEmitter emitter) throws IOException {
        User user = new User();
        user.setAge(RandomUtil.randomInt(100));
        user.setName("张三");
        // emitter.send(SseEmitter.event().id(UUID.randomUUID().toString()).data("哈哈哈哈哈："+ UUID.randomUUID()));
        emitter.send(user);
    }

    /**
     * 推送消息
     * @param message
     */
    private void send(String message){
        emitterMap.values().forEach(emitter -> {
            try {
                // doSend(emitter,message);
                doSend2(emitter,message);
            } catch (IOException e) {
                log.warn("客户端断开连接了");
            }
        });
    }

    /**
     * 推送消息
     * @param emitter
     * @param message
     * @throws IOException
     */
    private void doSend(SseEmitter emitter,String message) throws IOException {
        Set<ResponseBodyEmitter.DataWithMediaType> dataWithMediaTypes = SseEmitter.event()
                .id(UUID.randomUUID().toString())
                .name("message")
                .data(message)
                .build();
        emitter.send(dataWithMediaTypes);
    }

    private void doSend2(SseEmitter emitter,String message) throws IOException {
        String s = "e410d4c1d71c469b8d719de5d39783b7";
        emitter.send(SseEmitter.event().id(UUID.randomUUID().toString()).data(message));
    }
}

如果不使用SseEmitter类，咱们还可以扩展他，自定义一个类去继承SseEmitter类，同时链接接口，的返回类和链接容器都应该使用自定义类类似于如下
在这里插入图片描述

问题以及注意事项

如果业务服务是集群部署，轮询路由访问服务，避免内存消耗，重复缓存链接。
优化：Map<String, SseEmitter> emitterMap = new ConcurrentHashMap<>();
改为：Redis 方式。共享链接数据
服务环境采用Nginx进行反向代理，请求时一直peding：需要新增代理
location /connects/ {
proxy_buffering off; #关闭代理缓冲的
proxy_cache off;
proxy_read_timeout 18000s; #设置读取时间
proxy_send_timeout 18000s; #设置写入时间
proxy_http_version 1.1; # 设置http版本
proxy_pass http://xxxx-gateway-app:31001/; #路由地址
}
其中：
Nginx做反向代理时需要将proxy_buffering关闭
proxy_buffering off
如果经过了多重Nginx代理，还需要加上
加上响应头部x-accel-buffering，这样nginx就不会给后端响应数据加buffer
x-accel-buffering: no

推送最直接的方式就是使用第三推送平台，毕竟钱能解决的需求都不是问题，无需复杂的开发运维，直接可以使用，省时、省力、省心，像goEasy、极光推送都是很不错的三方服务商。
一般大型公司都有自研的消息推送平台，像我们本次实现的web站内信只是平台上的一个触点而已，短信、邮件、微信公众号、小程序凡是可以触达到用户的渠道都可以接入进来。
在这里插入图片描述

原文地址：https://blog.csdn.net/qq_43652793/article/details/143708885

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：DNS解析
下一篇：Logback 日志介绍及与Spring Boot 的整合【保姆级教程】

一款集成编码器的无刷电机驱动板
本实物模块从实物外观、组成原理及功能说明三部分来介绍这款集成编码器的无刷电机驱动板。
阅读更多2024-11-15
找不到Windows SDK 版本 10.0.22621.0
这意味着系统找不到，这是你构建项目所需要的 SDK 版本。
阅读更多2024-11-15
python制作一个简单的端口扫描器，用于检测目标主机上指定端口的开放状态
创建一个 socket 对象，AF_INET 表示使用 IPv4，SOCK_STREAM 表示使用 TCP。print("请正确指定目标端口，使用 -p 参数输入目标端口信息（多个端口用逗号
阅读更多2024-11-15
Java 实现鼠标单击右键弹出菜单项
在界面的任意地方右键点击出现一个菜单项 "Intro"，并且点击该项后弹出一个NewWindow。
阅读更多2024-11-15
dynamodb——事务
目的：提供在同一区域内对跨表的数据项执行原子和可序列化操作的能力，同时保证性能的可预测性，并且因此，没有采用传统意义上的交互式事务而是引入了两个新的单请求操作实现，实现从一致的快照中检索多个条目，
阅读更多2024-11-15
拆解测试显示Mac Mini (2024)固态硬盘并未锁定互换硬盘后仍可使用
当然那这里又会存在一个新坑：某些二手交易平台上价格相对较低的 Mac Mini (2024) 16GB+2TB 版有可能就是替换的硬盘，如果用户不了解这种情况的话可能就会被割韭菜。不过知名拆解网站 i
阅读更多2024-11-15
Ubuntu 24.04 安装 JDK 21
Ubuntu 24.04 安装 JDK 21
阅读更多2024-11-15
讲解C语言关键字
C语言是一种结构化的编程语言，它有一组特殊的保留字，称为“关键字”。这些关键字在程序中具有特定的含义和用途，不能用作标识符（例如变量名、函数名等）。
阅读更多2024-11-15
git入门环境搭建
git官网地址：https://git-scm.com/然后一路next 默认路径默认勾选就行。如果没有魔法的话，官网这个地址能卡死你。今天就写到这吧，11点多了该睡了，，，这里给个国内的git
阅读更多2024-11-15
定时器简介
在第一部分,我们主要讲的是定时器基本定时的功能，也就是定一个时间，然后让定时器每隔这个时间产生一个中断，来实现每隔一个固定时间执行一段程序的目的，比如你要做个时钟、秒表，或者使用一些程序算法的时候，都
阅读更多2024-11-15