java通过SSE实现消息推送

一、什么是SSE
严格地说，HTTP 协议无法做到服务器主动推送信息。但是，有一种变通方法，就是服务器向客户端声明，接下来要发送的是流信息（streaming）。
也就是说，发送的不是一次性的数据包，而是一个数据流，会连续不停地发送过来。这时，客户端不会关闭毗连，会一直等着服务器发过来的新的数据流，视频播放就是这样的例子。本质上，这种通讯就是以流信息的方式，完成一次用时很长的下载。
SSE 就是利用这种机制，利用流信息向浏览器推送信息。它基于 HTTP 协议，现在除了 IE，其他浏览器都支持。
二、怎样在java中实现SSE
在Spring Boot项目中，无需额外引入特定的依靠，由于Spring Web MVC模块已经内置了对SSE的支持。
1、编写SSE服务，来进行创建链接和发送消息

1. package com.smart3dmap.prjpntmanage.service;
3. import lombok.extern.slf4j.Slf4j;
4. import org.apache.commons.lang3.StringUtils;
5. import org.springframework.stereotype.Service;
6. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
8. import java.io.IOException;
9. import java.util.Map;
10. import java.util.concurrent.ConcurrentHashMap;
12. @Slf4j
13. @Service
14. public class SSEService {
15.     private static final Map<String,SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();
16.     public SseEmitter crateSse(String uid) {
17.         SseEmitter sseEmitter = new SseEmitter(0L);
18.         sseEmitter.onCompletion(() -> {
19.             log.info("[{}]结束链接" , uid);
20.             sseEmitterMap.remove(uid);
21.         });
22.         sseEmitter.onTimeout(() -> {
23.             log.info("[{}]链接超时",uid);
24.         });
25.         sseEmitter.onError(throwable -> {
26.             try{
27.                 log.info("[{}]链接异常，{}",uid,throwable.toString());
28.                 sseEmitter.send(SseEmitter.event()
29.                         .id(uid)
30.                         .name("发生异常")
31.                         .data("发生异常请重试")
32.                         .reconnectTime(3000));
33.                 sseEmitterMap.put(uid,sseEmitter);
34.             }catch (IOException e){
35.                 e.printStackTrace();
36.             }
37.         });
38.         try{
39.             sseEmitter.send(SseEmitter.event().reconnectTime(5000));
40.         }catch (IOException e){
41.             e.printStackTrace();
42.         }
43.         sseEmitterMap.put(uid,sseEmitter);
44.         log.info("[{}]创建sse连接成功!",uid);
45.         return sseEmitter;
46.     }
48.     public boolean sendMessage(String uid,String messageId,String message){
49.         if(StringUtils.isEmpty(message)){
50.             log.info("[{}]参数异常，msg为空",uid);
51.             return false;
52.         }
53.         SseEmitter sseEmitter = sseEmitterMap.get(uid);
54.         if(sseEmitter == null){
55.             log.info("[{}]sse连接不存在",uid);
56.             return  false;
57.         }
58.         try{
59.             sseEmitter.send(SseEmitter.event().id(messageId).reconnectTime(60000).data(message));
60.             log.info("用户{},消息ID：{}，推送成功：{}",uid,messageId,message);
61.             return true;
62.         }catch (IOException e){
63.             sseEmitterMap.remove(uid);
64.             log.info("用户{},消息ID：{}，消息推送失败：{}",uid,messageId,message);
65.             sseEmitter.complete();
66.             return false;
67.         }
68.     }
70.     public void closeSse(String uid){
71.         if(sseEmitterMap.containsKey(uid)){
72.             SseEmitter sseEmitter = sseEmitterMap.get(uid);
73.             sseEmitter.complete();
74.             sseEmitterMap.remove(uid);
75.         }else {
76.             log.info("用户{}连接已关闭",uid);
77.         }
78.     }
80. }

复制代码

2、编写对应的Controller来实现具体业务

1. package com.test.controller;
3. import cn.hutool.core.util.IdUtil;
4. import com.test.service.SSEService;
5. import org.springframework.beans.factory.annotation.Autowired;
6. import org.springframework.stereotype.Controller;
7. import org.springframework.web.bind.annotation.*;
8. import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
10. @Controller
11. @RequestMapping("test")
12. public class TestController {
13.     @Autowired
14.     private SSEService sseService;
15.     @GetMapping("createSse")
16.     @CrossOrigin
17.     public SseEmitter createSse(String uid)
18.     {
19.         return sseService.crateSse(uid);
20.     }
22.     @GetMapping("sendMsg")
23.     @ResponseBody
24.     @CrossOrigin
25.     public String sseChat(String uid){
26.         for (int i = 0; i < 10; i++) {
27.             sseService.sendMessage(uid,"消息"+i,IdUtil.fastUUID().replace("-",""));
28.         }
29.         return "OK";
30.     }
31.     @GetMapping("closeSse")
32.     @CrossOrigin
33.     public void closeSse(String uid){
34.         sseService.closeSse(uid);
35.     }
36. }

复制代码

三、前端实现消息的监听

1. <!doctype html>
2. <html lang="en">
3. <head>
4.     <meta charset="UTF-8">
5.     <meta name="viewport" content="width=device-width, initial-scale=1">
6.     <title>SSE消息推送监听</title>
7. </head>
8. <body>
9.    
11. </body>
12. </html>

复制代码

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。