devbox
我家小花儿
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Springboot之整合SSE实现消息推送_springboot sse推送

作者：我家小花儿 | 2024-03-29 11:28:20

赞

踩

springboot sse推送

前言

项目中涉及到部分请求，后端处理时间较长，使用常规Http请求，页面等待时间太长，对用户不友好，故考虑使用长链接进行消息推送，可选方案有WebSocket、SSE，WebSocket可实现双工通信，SSE仅支持服务端向客户端推送消息，根据实际使用场景，SSE即可满足，故选用SSE。

一、SSE是什么？

SSE技术是基于单工通信模式，只是单纯的客户端向服务端发送请求，服务端不会主动发送给客户端。服务端采取的策略是抓住这个请求不放，等数据更新的时候才返回给客户端，当客户端接收到消息后，再向服务端发送请求，周而复始。

  • 注意：因为EventSource对象是SSE的客户端，可能会有浏览器对其不支持，但谷歌、火狐、360是可以的，IE不可以。
  • 优点：SSE和WebSocket相比，最大的优势是便利，服务端不需要其他的类库，开发难度较低，SSE和轮询相比它不用处理很多请求，不用每次建立新连接，延迟较低。
  • 缺点：如果客户端有很多，那就要保持很多长连接，这会占用服务器大量内存和连接数
  • sse 规范：在 html5 的定义中，服务端 sse，一般需要遵循以下要求：
    Content-Type: text/event-stream;
    charset=UTF-8Cache-Control: no-cache
    Connection: keep-alive

二、使用步骤

1.客户端代码示例

  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>SseEmitter</title>
  6. </head>
  7. <body>
  8. <button onclick="closeSse()">关闭连接</button>
  9. <div id="message"></div>
  10. </body>
  11. <script>
  12.     let source = null;
  13.  
  14.     // 用时间戳模拟登录用户
  15.     const userId = new Date().getTime();
  16.  
  17.     if (!!window.EventSource) {
  18.  
  19.         // 建立连接
  20.         source = new EventSource('http://ip:端口/CreateSseConnect?clientId=39bd662b7942418595c21a1ef0af7fad');
  21.  
  22.         /**
  23.          * 连接一旦建立，就会触发open事件
  24.          * 另一种写法：source.onopen = function (event) {}
  25.          */
  26.         source.addEventListener('open', function (e) {
  27.             setMessageInnerHTML("建立连接。。。");
  28.         }, false);
  29.  
  30.         /**
  31.          * 客户端收到服务器发来的数据
  32.          * 另一种写法：source.onmessage = function (event) {}
  33.          */
  34.         source.addEventListener('message', function (e) {
  35.             setMessageInnerHTML(e.data);
  36.         });
  37.  
  38.  
  39.         /**
  40.          * 如果发生通信错误（比如连接中断），就会触发error事件
  41.          * 或者：
  42.          * 另一种写法：source.onerror = function (event) {}
  43.          */
  44.         source.addEventListener('error', function (e) {
  45.             if (e.readyState === EventSource.CLOSED) {
  46.                 setMessageInnerHTML("连接关闭");
  47.             } else {
  48.                 console.log(e);
  49.             }
  50.         }, false);
  51.  
  52.     } else {
  53.         setMessageInnerHTML("你的浏览器不支持SSE");
  54.     }
  55.  
  56.     // 监听窗口关闭事件，主动去关闭sse连接，如果服务端设置永不过期，浏览器关闭后手动清理服务端数据
  57.     window.onbeforeunload = function () {
  58.         closeSse();
  59.     };
  60.  
  61.     // 关闭Sse连接
  62.     function closeSse() {
  63.         source.close();
  64.         const httpRequest = new XMLHttpRequest();
  65.         httpRequest.open('GET', 'http://localhost:8080/sse/CloseConnect/?clientId=e410d4c1d71c469b8d719de5d39783b7', true);
  66.         httpRequest.send();
  67.         console.log("close");
  68.     }
  69.  
  70.     // 将消息显示在网页上
  71.     function setMessageInnerHTML(innerHTML) {
  72.         document.getElementById('message').innerHTML += innerHTML + '<br/>';
  73.     }
  74. </script>
  75. </html>

2.服务端整合

Controller：

  1. /**
  2.  * SSE长链接
  3.  */
  4. @RestController
  5. @RequestMapping("/sse")
  6. public class SseEmitterController {
  7.  
  8.     @Autowired
  9.     private SseEmitterService sseEmitterService;
  10.  
  11.     /**
  12.      * 创建SSE长链接
  13.      *
  14.      * @param clientId   客户端唯一ID(如果为空，则由后端生成并返回给前端)
  15.      * @return org.springframework.web.servlet.mvc.method.annotation.SseEmitter
  16.      **/
  17.     @CrossOrigin //如果nginx做了跨域处理，此处可去掉
  18.     @GetMapping("/CreateSseConnect")
  19.     public SseEmitter createSseConnect(@RequestParam(name = "clientId", required = false) String clientId) {
  20.         return sseEmitterService.createSseConnect(clientId);
  21.     }
  22.  
  23.     /**
  24.      * 关闭SSE连接
  25.      *
  26.      * @param clientId 客户端ID
  27.      **/
  28.     @GetMapping("/CloseSseConnect")
  29.     public Result closeSseConnect(String clientId) {
  30.         sseEmitterService.closeSseConnect(clientId);
  31.         return ResultGenerator.genSuccessResult(true);
  32.     }
  33.  
  34. }

ServiceImpl

  1. @Service
  2. public class SseEmitterServiceImpl implements SseEmitterService {
  3. 	
  4.     /**
  5.      * 容器，保存连接，用于输出返回
  6.      */
  7.     private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();
  8.     
  9.     @Override
  10.     public SseEmitter createSseConnect(String clientId) {
  11.         // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
  12.         SseEmitter sseEmitter = new SseEmitter(0L);
  13.         // 是否需要给客户端推送ID
  14.         if (StringUtils.isBlank(clientId)) {
  15.             clientId = IdUtil.simpleUUID();
  16.         }
  17.         // 注册回调
  18.         sseEmitter.onCompletion(completionCallBack(clientId));
  19.         sseCache.put(clientId, sseEmitter);
  20.         logger.info("创建新的sse连接，当前用户：{}", clientId);
  21.  
  22.         try {
  23.             sseEmitter.send(SseEmitter.event().id(SseEmitterConstant.CLIENT_ID).data(clientId));
  24.         } catch (IOException e) {
  25.             logger.error("SseEmitterServiceImpl[createSseConnect]: 创建长链接异常，客户端ID:{}", clientId, e);
  26.             throw new BusinessException("创建连接异常！", e);
  27.         }
  28.         return sseEmitter;
  29.     }
  30.  
  31.     @Override
  32.     public void closeSseConnect(String clientId) {
  33.         SseEmitter sseEmitter = sseCache.get(clientId);
  34.         if (sseEmitter != null) {
  35.             sseEmitter.complete();
  36.             removeUser(clientId);
  37.         }
  38.     }
  39.  
  40. 	// 根据客户端id获取SseEmitter对象
  41.     @Override
  42.     public SseEmitter getSseEmitterByClientId(String clientId) {
  43.         return sseCache.get(clientId);
  44.     }
  45.  
  46. 	// 推送消息到客户端，此处结合业务代码，业务中需要推送消息处调用即可向客户端主动推送消息
  47.     @Override
  48.     public void sendMsgToClient(List<SseEmitterResultVO> sseEmitterResultVOList) {
  49.         if (CollectionUtil.isEmpty(sseCache)) {
  50.             return;
  51.         }
  52.         for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
  53.             sendMsgToClientByClientId(entry.getKey(), sseEmitterResultVOList, entry.getValue());
  54.         }
  55.     }
  56.  
  57.     /**
  58.      * 推送消息到客户端
  59.      * 此处做了推送失败后，重试推送机制，可根据自己业务进行修改
  60.      *
  61.      * @param clientId               客户端ID
  62.      * @param sseEmitterResultVOList 推送信息，此处结合具体业务，定义自己的返回值即可
  63.      **/
  64.     private void sendMsgToClientByClientId(String clientId, List<SseEmitterResultVO> sseEmitterResultVOList, SseEmitter sseEmitter) {
  65.         if (sseEmitter == null) {
  66.             logger.error("SseEmitterServiceImpl[sendMsgToClient]: 推送消息失败：客户端{}未创建长链接,失败消息:{}",
  67.                     clientId, sseEmitterResultVOList.toString());
  68.             return;
  69.         }
  70.  
  71.         SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(SseEmitterConstant.TASK_RESULT).data(sseEmitterResultVOList, MediaType.APPLICATION_JSON);
  72.         try {
  73.             sseEmitter.send(sendData);
  74.         } catch (IOException e) {
  75.             // 推送消息失败，记录错误日志，进行重推
  76.             logger.error("SseEmitterServiceImpl[sendMsgToClient]: 推送消息失败：{},尝试进行重推", sseEmitterResultVOList.toString(), e);
  77.             boolean isSuccess = true;
  78.             // 推送消息失败后，每隔10s推送一次，推送5次
  79.             for (int i = 0; i < 5; i++) {
  80.                 try {
  81.                     Thread.sleep(10000);
  82.                     sseEmitter = sseCache.get(clientId);
  83.                     if (sseEmitter == null) {
  84.                         logger.error("SseEmitterServiceImpl[sendMsgToClient]：{}的第{}次消息重推失败，未创建长链接", clientId, i + 1);
  85.                         continue;
  86.                     }
  87.                     sseEmitter.send(sendData);
  88.                 } catch (Exception ex) {
  89.                     logger.error("SseEmitterServiceImpl[sendMsgToClient]：{}的第{}次消息重推失败", clientId, i + 1, ex);
  90.                     continue;
  91.                 }
  92.                 logger.info("SseEmitterServiceImpl[sendMsgToClient]：{}的第{}次消息重推成功,{}", clientId, i + 1, sseEmitterResultVOList.toString());
  93.                 return;
  94.             }
  95.         }
  96.     }
  97.  
  98.     /**
  99.      * 长链接完成后回调接口(即关闭连接时调用)
  100.      *
  101.      * @param clientId 客户端ID
  102.      * @return java.lang.Runnable
  103.      **/
  104.     private Runnable completionCallBack(String clientId) {
  105.         return () -> {
  106.             logger.info("结束连接：{}", clientId);
  107.             removeUser(clientId);
  108.         };
  109.     }
  110.  
  111.     /**
  112.      * 连接超时时调用
  113.      *
  114.      * @param clientId 客户端ID
  115.      * @return java.lang.Runnable
  116.      **/
  117.     private Runnable timeoutCallBack(String clientId) {
  118.         return () -> {
  119.             logger.info("连接超时：{}", clientId);
  120.             removeUser(clientId);
  121.         };
  122.     }
  123.  
  124.     /**
  125.      * 推送消息异常时，回调方法
  126.      *
  127.      * @param clientId 客户端ID
  128.      * @return java.util.function.Consumer<java.lang.Throwable>
  129.      **/
  130.     private Consumer<Throwable> errorCallBack(String clientId) {
  131.         return throwable -> {
  132.             logger.error("SseEmitterServiceImpl[errorCallBack]：连接异常,客户端ID:{}", clientId);
  133.  
  134.             // 推送消息失败后，每隔10s推送一次，推送5次
  135.             for (int i = 0; i < 5; i++) {
  136.                 try {
  137.                     Thread.sleep(10000);
  138.                     SseEmitter sseEmitter = sseCache.get(clientId);
  139.                     if (sseEmitter == null) {
  140.                         logger.error("SseEmitterServiceImpl[errorCallBack]：第{}次消息重推失败,未获取到 {} 对应的长链接", i + 1, clientId);
  141.                         continue;
  142.                     }
  143.                     sseEmitter.send("失败后重新推送");
  144.                 } catch (Exception e) {
  145.                     e.printStackTrace();
  146.                 }
  147.             }
  148.         };
  149.     }
  150.  
  151.     /**
  152.      * 移除用户连接
  153.      *
  154.      * @param clientId 客户端ID
  155.      **/
  156.     private void removeUser(String clientId) {
  157.         sseCache.remove(clientId);
  158.         logger.info("SseEmitterServiceImpl[removeUser]:移除用户：{}", clientId);
  159.     }
  160. }

3. Nginx配置

如果项目中使用nginx对后端服务做了代理，nginx代理转发后，默认会在1min的时候断掉长链接，SSE需要设置自己的长链接时间，则需要在nginx中进行配置；
在反向代理的location块中加入如下配置

  1. proxy_set_header Host $http_host;  ##proxy_set_header用来重定义发往后端服务器的请求头
  2. proxy_set_header X-Real-IP $remote_addr;
  3. proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
  4. proxy_set_header X-Forwarded-Proto $scheme;
  5. proxy_buffering off;
  6. proxy_http_version  1.1;
  7. proxy_read_timeout 600s; ##设置SSE长链接保持时间为 600s

4. 请求示例

在这里插入图片描述
在这里插入图片描述

常见问题

1、前端报错：EventSource’s response has a MIME type (“application/json”) that is not “text/event-stream”. Aborting the connection
前端在创建SSE长链接时，完整的请求(包括参数和参数值)都必须放在new EventSource(完整请求)中；
2、创建长链接时，接口状态一直处于pending，检查后端nginx是否做相应配置；
3、推送消息失败：检查客户端创建链接时的id，和推送消息时的id是否一致；

总结

整体业务流程为：客户端创建链接——>服务端保持生成SseEmitter对象，并通过SseEmitter对象实现向客户端主动推送消息——>客户端收到推送消息后，刷新页面(根据推送消息，请求相关业务接口)

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号