Springboot 集成 SSE 向前端推送消息

旧城等待， 2023-09-29 13:27 133阅读 0赞

#### Sse推送 ####

*   *  Sse介绍
     *  特点分析
     *  应用场景
     *  Spring Boot 集成
     *  测试
     *  前端代码

### Sse介绍 ###

sse(Server Sent Event)，直译为服务器发送事件，顾名思义，也就是客户端可以获取到服务器发送的事件

我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕；但是在 sse 的场景下，客户端发起请求，连接一直保持，服务端有数据就可以返回数据给客户端，这个返回可以是多次间隔的方式

### 特点分析 ###

SSE 最大的特点，可以简单规划为两个

*  长连接
 *  服务端可以向客户端推送信息

了解 websocket 的小伙伴，可能也知道它也是长连接，可以推送信息，但是它们有一个明显的区别

sse 是单通道，只能服务端向客户端发消息；而 webscoket 是双通道  
那么为什么有了 webscoket 还要搞出一个 sse 呢？既然存在，必然有着它的优越之处

<table> 
 <thead> 
  <tr> 
   <th>sse</th> 
   <th>websocket</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>http 协议</td> 
   <td>独立的 websocket 协议</td> 
  </tr> 
  <tr> 
   <td>轻量，使用简单</td> 
   <td>相对复杂</td> 
  </tr> 
  <tr> 
   <td>默认支持断线重连</td> 
   <td>需要自己实现断线重连</td> 
  </tr> 
  <tr> 
   <td>文本传输</td> 
   <td>二进制传输</td> 
  </tr> 
  <tr> 
   <td>支持自定义发送的消息类型</td> 
   <td>-</td> 
  </tr> 
 </tbody> 
</table>

### 应用场景 ###

从 sse 的特点出发，我们可以大致的判断出它的应用场景，需要轮询获取服务端最新数据的 case 下，多半是可以用它的

比如显示当前网站在线的实时人数，法币汇率显示当前实时汇率，电商大促的实时成交额等等…

### Spring Boot 集成 ###

**项目结构**  
![在这里插入图片描述][2910d8bb127346648c9797b7aae38aa8.png]

**使用到的依赖**

<dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>cn.hutool</groupId>
        <artifactId>hutool-all</artifactId>
        <version>5.7.16</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>

**消息实体**

package com.ddz.sse.entity;
    
    import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    
    /**
     * 消息体
     *
     * @author Lenovo
     * @date 2022/5/6
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class MessageVo {
        
        /**
         * 客户端id
         */
        private String clientId;
        /**
         * 传输数据体(json)
         */
        private String data;
    }

**接口**

package com.ddz.sse.service;
    
    import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    
    /**
     * @author Lenovo
     * @date 2022/5/6
     */
    public interface SseEmitterService {
        
        /**
         * 创建连接
         *
         * @param clientId 客户端ID
         */
        SseEmitter createConnect(String clientId);
    
        /**
         * 根据客户端id获取SseEmitter对象
         *
         * @param clientId 客户端ID
         */
        SseEmitter getSseEmitterByClientId(String clientId);
    
        /**
         * 发送消息给所有客户端
         *
         * @param msg 消息内容
         */
        void sendMessageToAllClient(String msg);
    
        /**
         * 给指定客户端发送消息
         *
         * @param clientId 客户端ID
         * @param msg      消息内容
         */
        void sendMessageToOneClient(String clientId, String msg);
    
        /**
         * 关闭连接
         *
         * @param clientId 客户端ID
         */
        void closeConnect(String clientId);
    }

**实现类**

package com.ddz.sse.service.impl;
    
    import cn.hutool.core.map.MapUtil;
    import cn.hutool.core.util.IdUtil;
    import cn.hutool.core.util.StrUtil;
    import cn.hutool.http.HttpStatus;
    import com.ddz.sse.entity.MessageVo;
    import com.ddz.sse.service.SseEmitterService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.http.MediaType;
    import org.springframework.stereotype.Service;
    import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    
    import java.io.IOException;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.function.Consumer;
    
    @Slf4j
    @Service
    public class SseEmitterServiceImpl implements SseEmitterService {
        
        /**
         * 容器，保存连接，用于输出返回 ;可使用其他方法实现
         */
        private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
    
    
        /**
         * 根据客户端id获取SseEmitter对象
         *
         * @param clientId 客户端ID
         */
        @Override
        public SseEmitter getSseEmitterByClientId(String clientId) {
        
            return sseCache.get(clientId);
        }
    
    
        /**
         * 创建连接
         *
         * @param clientId 客户端ID
         */
        @Override
        public SseEmitter createConnect(String clientId) {
        
            // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
            SseEmitter sseEmitter = new SseEmitter(0L);
            // 是否需要给客户端推送ID
            if (StrUtil.isBlank(clientId)) {
        
                clientId = IdUtil.simpleUUID();
            }
            // 注册回调
            sseEmitter.onCompletion(completionCallBack(clientId));     // 长链接完成后回调接口(即关闭连接时调用)
            sseEmitter.onTimeout(timeoutCallBack(clientId));        // 连接超时回调
            sseEmitter.onError(errorCallBack(clientId));          // 推送消息异常时，回调方法
            sseCache.put(clientId, sseEmitter);
            log.info("创建新的sse连接，当前用户：{}    累计用户:{}", clientId, sseCache.size());
            try {
        
                // 注册成功返回用户信息
                sseEmitter.send(SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_CREATED)).data(clientId, MediaType.APPLICATION_JSON));
            } catch (IOException e) {
        
                log.error("创建长链接异常，客户端ID:{}   异常信息:{}", clientId, e.getMessage());
            }
            return sseEmitter;
        }
    
        /**
         * 发送消息给所有客户端
         *
         * @param msg 消息内容
         */
        @Override
        public void sendMessageToAllClient(String msg) {
        
            if (MapUtil.isEmpty(sseCache)) {
        
                return;
            }
            // 判断发送的消息是否为空
    
            for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
        
                MessageVo messageVo = new MessageVo();
                messageVo.setClientId(entry.getKey());
                messageVo.setData(msg);
                sendMsgToClientByClientId(entry.getKey(), messageVo, entry.getValue());
            }
    
        }
    
        /**
         * 给指定客户端发送消息
         *
         * @param clientId 客户端ID
         * @param msg      消息内容
         */
        @Override
        public void sendMessageToOneClient(String clientId, String msg) {
        
            MessageVo messageVo = new MessageVo(clientId, msg);
            sendMsgToClientByClientId(clientId, messageVo, sseCache.get(clientId));
        }
    
        /**
         * 关闭连接
         *
         * @param clientId 客户端ID
         */
        @Override
        public void closeConnect(String clientId) {
        
            SseEmitter sseEmitter = sseCache.get(clientId);
            if (sseEmitter != null) {
        
                sseEmitter.complete();
                removeUser(clientId);
            }
        }
    
        /**
         * 推送消息到客户端
         * 此处做了推送失败后，重试推送机制，可根据自己业务进行修改
         *
         * @param clientId  客户端ID
         * @param messageVo 推送信息，此处结合具体业务，定义自己的返回值即可
         **/
        private void sendMsgToClientByClientId(String clientId, MessageVo messageVo, SseEmitter sseEmitter) {
        
            if (sseEmitter == null) {
        
                log.error("推送消息失败：客户端{}未创建长链接,失败消息:{}",
                        clientId, messageVo.toString());
                return;
            }
            SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_OK))
                    .data(messageVo, MediaType.APPLICATION_JSON);
            try {
        
                sseEmitter.send(sendData);
            } catch (IOException e) {
        
                // 推送消息失败，记录错误日志，进行重推
                log.error("推送消息失败：{},尝试进行重推", messageVo.toString());
                boolean isSuccess = true;
                // 推送消息失败后，每隔10s推送一次，推送5次
                for (int i = 0; i < 5; i++) {
        
                    try {
        
                        Thread.sleep(10000);
                        sseEmitter = sseCache.get(clientId);
                        if (sseEmitter == null) {
        
                            log.error("{}的第{}次消息重推失败，未创建长链接", clientId, i + 1);
                            continue;
                        }
                        sseEmitter.send(sendData);
                    } catch (Exception ex) {
        
                        log.error("{}的第{}次消息重推失败", clientId, i + 1, ex);
                        continue;
                    }
                    log.info("{}的第{}次消息重推成功,{}", clientId, i + 1, messageVo.toString());
                    return;
                }
            }
        }
    
    
        /**
         * 长链接完成后回调接口(即关闭连接时调用)
         *
         * @param clientId 客户端ID
         **/
        private Runnable completionCallBack(String clientId) {
        
            return () -> {
        
                log.info("结束连接：{}", clientId);
                removeUser(clientId);
            };
        }
    
        /**
         * 连接超时时调用
         *
         * @param clientId 客户端ID
         **/
        private Runnable timeoutCallBack(String clientId) {
        
            return () -> {
        
                log.info("连接超时：{}", clientId);
                removeUser(clientId);
            };
        }
    
        /**
         * 推送消息异常时，回调方法
         *
         * @param clientId 客户端ID
         **/
        private Consumer<Throwable> errorCallBack(String clientId) {
        
            return throwable -> {
        
                log.error("SseEmitterServiceImpl[errorCallBack]：连接异常,客户端ID:{}", clientId);
    
                // 推送消息失败后，每隔10s推送一次，推送5次
                for (int i = 0; i < 5; i++) {
        
                    try {
        
                        Thread.sleep(10000);
                        SseEmitter sseEmitter = sseCache.get(clientId);
                        if (sseEmitter == null) {
        
                            log.error("SseEmitterServiceImpl[errorCallBack]：第{}次消息重推失败,未获取到 {} 对应的长链接", i + 1, clientId);
                            continue;
                        }
                        sseEmitter.send("失败后重新推送");
                    } catch (Exception e) {
        
                        e.printStackTrace();
                    }
                }
            };
        }
    
        /**
         * 移除用户连接
         *
         * @param clientId 客户端ID
         **/
        private void removeUser(String clientId) {
        
            sseCache.remove(clientId);
            log.info("SseEmitterServiceImpl[removeUser]:移除用户：{}", clientId);
        }
    }

**控制器类**

package com.ddz.sse.controller;
    
    import com.ddz.sse.entity.MessageVo;
    import com.ddz.sse.service.SseEmitterService;
    import org.springframework.web.bind.annotation.*;
    import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
    
    import javax.annotation.Resource;
    
    /**
     * SSE长链接
     */
    @RestController
    @RequestMapping("/sse")
    public class SseEmitterController {
        
    
        @Resource
        private SseEmitterService sseEmitterService;
    
    
        @CrossOrigin
        @GetMapping("/createConnect")
        public SseEmitter createConnect(String clientId) {
        
            return sseEmitterService.createConnect(clientId);
        }
    
        @CrossOrigin
        @PostMapping("/broadcast")
        public void sendMessageToAllClient(@RequestBody(required = false) String msg) {
        
            sseEmitterService.sendMessageToAllClient(msg);
        }
    
        @CrossOrigin
        @PostMapping("/sendMessage")
        public void sendMessageToOneClient(@RequestBody(required = false) MessageVo messageVo) {
        
            if (messageVo.getClientId().isEmpty()) {
        
                return;
            }
            sseEmitterService.sendMessageToOneClient(messageVo.getClientId(), messageVo.getData());
        }
    
        @CrossOrigin
        @GetMapping("/closeConnect")
        public void closeConnect(@RequestParam(required = true) String clientId) {
        
            sseEmitterService.closeConnect(clientId);
        }
    
    }

### 测试 ###

自己创建clientid  
![在这里插入图片描述][8451c27606e744f69a8e1211524e39c0.png]  
使用系统创建clientid  
![在这里插入图片描述][a6e7cdf555dd40bea82b9a406565fc80.png]  
这里浏览器一直在转圈说明是连接成功.  
**我们使用APIPost接口测试工具测试**  
![在这里插入图片描述][8397edc2fd7740d28a9d92a587862d40.png]  
这里ddz123的是能收到我们推送的消息的.

### 前端代码 ###

<!doctype html>
    <html lang="en">
    <head>
        <title>Sse测试文档</title>
    </head>
    <body>
    <div>sse测试</div>
    <div id="data"></div>
    </body>
    </html>
    <script>
        var source = new EventSource('http://192.168.1.128:8889/sse/createConnect?clientId=ddz');
        source.onmessage = function (event) {
          	
            text = document.getElementById('data').innerText;
            text += '\n' + event.data;
            document.getElementById('data').innerText = text;
        };
        
        source.onopen = function (event) {
          
            text = document.getElementById('data').innerText;
            text += '\n 开启: ';
            console.log(event);
            document.getElementById('data').innerText = text;
        };
    </script>

如果需要携带token请求,那就需要引入一个插件

npm install event-source-polyfill

我这里的是vue的代码;**代码不完善只供参考一下**

<template>
        <div>sse测试</div>
        <div id="data"></div>
    </template>
    
    <script>
    import {
          
        EventSourcePolyfill
    } from 'event-source-polyfill';
    
    export default {
          
        name: 'sse',
        data() {
          
            return {
          
                token:'获取到的令牌'
            }
        },
        mounted() {
           },
        created() {
          
            this.createContent('http://192.168.1.128:8080/syscon/sse/createConnect','?clientId=ddz',this.token)
        },
        methods: {
          
            //创建连接
            createContent(url, params, token) {
          
                return new EventSourcePolyfill(url, params, {
          
                    headers: {
          
                        Authorization: token
                    }
                })
            },
        },
    }
    </script>

文章参考于[https://www.cnblogs.com/yihuihui/p/12622729.html][https_www.cnblogs.com_yihuihui_p_12622729.html]

[2910d8bb127346648c9797b7aae38aa8.png]: https://img-blog.csdnimg.cn/2910d8bb127346648c9797b7aae38aa8.png
[8451c27606e744f69a8e1211524e39c0.png]: https://img-blog.csdnimg.cn/8451c27606e744f69a8e1211524e39c0.png
[a6e7cdf555dd40bea82b9a406565fc80.png]: https://img-blog.csdnimg.cn/a6e7cdf555dd40bea82b9a406565fc80.png
[8397edc2fd7740d28a9d92a587862d40.png]: https://img-blog.csdnimg.cn/8397edc2fd7740d28a9d92a587862d40.png
[https_www.cnblogs.com_yihuihui_p_12622729.html]: https://www.cnblogs.com/yihuihui/p/12622729.html