基于 SSE 实现服务端消息主动推送解决方案

Posted 2022-08-07 小毕超

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了基于 SSE 实现服务端消息主动推送解决方案相关的知识，希望对你有一定的参考价值。

一、SSE 服务端消息推送

SSE 是 Server-Sent Events 的简称，是一种服务器端到客户端(浏览器)的单项消息推送。对应的浏览器端实现 Event Source 接口被制定为html5 的一部分。不过现在IE不支持该技术，只能通过轮训的方式实现。相比于 WebSocket，SSE 简单很多，服务器端和客户端工作量都要小很多、简单很多，同时实现的功能也要有局限。

相比于 WebSocket 两者的区别：

WebSocket 是全双工通道，可以双向通信，功能更强。SSE是单向通道，只能服务器向浏览器端发送。
WebSocket 是一个新的协议，需要服务器端支持。SSE则是部署在 HTTP协议之上的，现有的服务器软件都支持。
SSE是一个轻量级协议，相对简单。WebSocket是一种较重的协议，相对复杂。
SSE默认支持断线重连，WebSocket则需要额外部署。
SSE支持自定义发送的数据类型。
SSE不支持CORS，参数 url 就是服务器网址，必须与当前网页的网址在同一个网域（domain），而且协议和端口都必须相同。

在我们平常使用 SpringBoot 进行开发中，其实已经集成好了 SSE ，里面有个 SseEmitter 类已经封装好了相关操作，可以方便的实现功能。

但是在实现的时候需要注意下是否要兼容 IE 浏览器的使用，IE 浏览器目前不支持 Event Source 接口，如果需要兼容 IE 可以创建一个顺序队列比如放在 Redis 中或消息队列中，IE 客户端通过接口轮训的方式每次都从队列中消费消息，以实现 SSE 的功能。

下面分别从服务端和客户端进行实施。

二、服务端

服务端需要考虑 IE 浏览器不支持的情况，对于 IE 可以通过轮训的方式实现，首先新建一个 SpringBoot 项目，声明 SseEmitter 连接：

@Slf4j
public class SseEmitterServer 

    private static AtomicInteger count = new AtomicInteger(0);

    private static Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    public static SseEmitter connect(String userId) 
        SseEmitter sseEmitter = new SseEmitter(0L);
        sseEmitter.onCompletion(() -> 
            log.info("结束连接：", userId);
            removeUser(userId);
        );
        sseEmitter.onError(throwable -> 
            log.info("连接异常：", userId);
            removeUser(userId);
        );
        sseEmitter.onTimeout(() -> 
            log.info("连接超时：", userId);
            removeUser(userId);
        );
        sseEmitterMap.put(userId, sseEmitter);
        count.getAndIncrement();
        log.info("创建新的sse连接，当前用户：", userId);
        return sseEmitter;
    


    public static void sendMessage(String userId, Object message) 
        if (sseEmitterMap.containsKey(userId)) 
            try 
                sseEmitterMap.get(userId).send(message);
                log.info("SSE 发送信息成功！id =  , message:  ", userId, message);
             catch (IOException e) 
                log.error("[]推送异常:", userId, e.getMessage());
                removeUser(userId);
            
         else 
            log.warn("SSE 发送信息异常，用户不存在：id =  ", userId);
        
    
    
    private static void removeUser(String userId) 
        sseEmitterMap.remove(userId);
        count.getAndDecrement();

然后声明 SSEWebServer 释放对客户端的接口，对于 IE 的话我们就直接提供一个接口，每次都从队列中消费消息，这里以 LinkedBlockingDeque 为例实现一个单机的队列，如果是分布式的可能就要考虑 Redis 或消息队列：

@Slf4j
@CrossOrigin
@RestController
@RequestMapping("/sse")
public class SSEWebServer 

    private static Cache<String, LinkedBlockingDeque<SseEvent>> sseCache = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(60, TimeUnit.MINUTES)
            .build();

    /* *
     * sse 连接服务
     */
    @GetMapping("/sseEvent/userId")
    public SseEmitter push(@PathVariable("userId") String userId) 
        return SseEmitterServer.connect(userId);
    

    //IE 浏览器不支持SSE 采用轮训
    @GetMapping("/sseEventIE/userId")
    public ResponseEntity pushIe(@PathVariable("userId") String userId) 
        if (StringUtils.isEmpty(userId)) 
            return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(" userId is Empty ! ");
        
        log.info("IE 连接，userId =  ", userId);
        try 
            SseEvent poll = Objects.requireNonNull(sseCache.getIfPresent(userId)).poll();
            return poll == null ? ResponseEntity.status(HttpStatus.BAD_REQUEST).body("连接失败！") : ResponseEntity.ok().body(poll.getMsg());
         catch (Exception e) 
            return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(e.getMessage());
        
    

    static boolean publicMsg(SseEvent event) 
        LinkedBlockingDeque<SseEvent> ifPresent = sseCache.getIfPresent(event.getUserId());
        if (ifPresent == null) 
            sseCache.put(event.getUserId(), new LinkedBlockingDeque<SseEvent>());
        
        log.info("添加到队列，userId： ", event.getUserId());
        return Objects.requireNonNull(sseCache.getIfPresent(event.getUserId())).offer(event);

上面为考虑 IE 的兼容性，多增加了一个接口和队列，因此在发布的时候，就需要同时向 SSE 和队列抛数据，因此这块可以在增加一个事件发布：

事件发布我们就使用 Spring 自带的 ApplicationListener 来实现。

首先创建一个事件交易类：

@Getter
@Setter
@ToString
public class SseEvent<T> extends ApplicationEvent 
    private int code;
    private String userId;
    private T msg;

    public SseEvent(Object source) 
        super(source);

声明事件监听，在这里同时向 SSE 和队列发送消息：

@Slf4j
@Component
public class SseListener implements ApplicationListener<SseEvent> 
    @Override
    public void onApplicationEvent(SseEvent event) 
        SseEmitterServer.sendMessage(event.getUserId(), event.getMsg());
        SSEWebServer.publicMsg(event);

最后再创建一个测试接口，便于我们下面的测试：

@RestController
public class TestController 

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;


    @GetMapping("/test/userId/message")
    public ResponseEntity test(@PathVariable("userId") String userId, @PathVariable("message") String message) 
        SseEvent<String> sseEvent = new SseEvent<>(this);
        sseEvent.setCode(200);
        sseEvent.setMsg(message);
        sseEvent.setUserId(userId);
        applicationEventPublisher.publishEvent(sseEvent);
        return ResponseEntity.ok().build();

到此服务端就已经完成了，下面开始客户端的对接：

三、客户端

相比于服务端，客户端就显得非常简单了，但也要考虑 IE 不支持的情况，需要进行判断，如果是 IE 的话，就换成轮训的方式：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

</body>
<script src="js/jquery-1.10.2.min.js"></script>
<script>

    openSSE("1122", function (msg) 
        console.log("收到服务器推送消息：" + msg);
    );

    function openSSE(userId, callback) 
        if (window.EventSource) 
            var source = new EventSource('http://localhost:8080/sse/sseEvent/' + userId);
            source.onmessage = function (event) 
                callback(event.data);
            ;
         else 
            //ie 不支持sse 采用轮训
            window.setInterval(function () 
                $.ajax(
                    url: "http://localhost:8080/sse/sseEventIE/" + userId + "?" + new Date().getTime(),
                    method: "GET",
                    success: function (result) 
                        callback(result);
                    
                )
            , 1000);
        
    

</script>
</html>