SSE实时数据推送

创建SSE连接对象后可以实时的根据信息对信息进行推送。一般在系统中我们会采用Map存储用户的信息。

// 5. 创建SSE连接，设置超时时间为1小时 SseEmitter emitter = new SseEmitter(60 * 60 * 1000L); //如果创建时时间设置为0L表示改连接永不超时只能通过监听器删除或者通过用户手动删除

当通过Sse连接到客户端时可以通过不同的API确保系统的正常使用。通过在API中设置Sse的策略，自由配置当Sse出现异常时的策略。

emitter.onCompletion(() -> { emitters.remove(connectionKey); System.out.println("SSE连接完成，用户: " + userName); }); //如果连接超时则调用改内部方法 emitter.onTimeout(() -> { emitters.remove(connectionKey); System.out.println("SSE连接超时，用户: " + userName); }); //Sse连接发生异常后调用内部方法 emitter.onError((ex) -> { emitters.remove(connectionKey); System.out.println("SSE连接错误，用户: " + userName + ", 错误: " + ex.getMessage()); });

通过调用SseEmitter对象的send方法向客户端进行发送数据，调用completeWithError可以进行关闭连接。关闭连接时要进行查看Map中存储的连接是否被删除防止因为未删除而造成无效连接的资源占用。

通过sse连接循环发送不同的数据信息：

循环发送数据我们可以既可以通过创建实时调度线程池进行实现也可以通过springboot中的任务调度注解搭配SseEmitter的reconnectTime方法进行循环发送数据。

任务调度线程池：

// 启动定时任务，每30秒推送一次数据 scheduler.scheduleAtFixedRate(this::broadcastData, 0, REFRESH_INTERVAL, TimeUnit.SECONDS);

通过springboot中的任务调度搭配SseEmitter的reconnectTime方法进行循环发送数据：

@Scheduled(cron = "0/30 * * * * *")//间隔30秒发送一次数据 // 发送连接成功事件，发送成功后间隔1秒才可以再次发送 SseEmitter.event().reconnectTime(1000);

Sse连接心跳检测：

当我们创建了永久不超时的SSe连接后如果客户端并未及时关闭或长时间占用系统资源我们可以通过创建心跳检测的心跳检测任务进行移除该连接

@Scheduled(cron = "0/30 * * * * *") public void SseHeartbeatCheck() { if (emitters.size() > 0){ //记录要删除的连接 List<String> toRemove = new ArrayList<>(); long currentTime = System.currentTimeMillis(); for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) { SseEmitter emitter = entry.getValue(); try { //心跳检测包 HashMap<String, Object> heart = new HashMap<>(); heart.put("type", "heartbeat"); heart.put("timestamp", currentTime); heart.put("serverTime", new java.util.Date()); SseEmitter.SseEventBuilder heartbeat = SseEmitter.event() .name("heartbeat") .id(String.valueOf(currentTime)) .data(heart); emitter.send(heartbeat); }catch (Exception e){ //移除异常连接 toRemove.add(entry.getKey()); } } //批量移除异常连接 toRemove.stream().forEach(E -> { SseEmitter emitter = emitters.get(E); if (emitter != null) { emitter.complete(); emitters.remove(emitter); } }); } }