Sse实现服务端长时间推送消息到客户端(H5)

SseEmitter是SpringMVC(4.2+)提供的一种技术,它是基于Http协议的，相比WebSocket，它更轻量，但是它只能从服务端向客户端单向发送信息。而webscoket 是双通道, 在SpringBoot中我们无需引用其他jar就可以使用。

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

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

websocket 协议，使用相对复杂默认支持断线重连需要自己实现断线重连文本传输二进制传输支持自定义发送的消息类型

并且H5页面可以在服务端挂了的情况下, 自动不断尝试重连接。

在 html5 的定义中，服务端 sse，一般需要遵循以下要求

请求头

开启长连接 + 流方式传递

@RestController
public class TemperatureController implements InitializingBean {

    static final long SSE_SESSION_TIMEOUT = 30 * 60 * 1000L;

    private static final Logger log = LoggerFactory.getLogger(TemperatureController.class);

    private final Set clients = new CopyOnWriteArraySet<>();

    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(4);

    @CrossOrigin
    @GetMapping(value = "/map-stream")
    public SseEmitter events(HttpServletRequest request) {
        log.info("SSE stream opened for client: (为客户端打开SSE流)" + request.getRemoteAddr());
        //默认为30000毫秒 0L为不超时 超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter emitter = new SseEmitter(SSE_SESSION_TIMEOUT); 
        clients.add(emitter);
        // Remove SseEmitter from active clients on error or client disconnect
        emitter.onTimeout(() -> { //  超时了回调删除连接
            log.error("SseEmitter-1, 客户端连接超时");
            clients.remove(emitter);
        });
        emitter.onCompletion(() -> { // 完成了断开连接回调
            log.error("SseEmitter-2, 已完成");
            clients.remove(emitter);
        });
        emitter.onError((throwable)->{ // 错误时
            log.error("SseEmitter-3", throwable);
            clients.remove(emitter);
        });
        return emitter;
    }
    
    @ExceptionHandler(value = AsyncRequestTimeoutException.class)
    public ModelAndView handleTimeout(HttpServletResponse rsp) throws IOException {
        if (!rsp.isCommitted()) {
            rsp.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
        }
        return new ModelAndView();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        final int[] i = {0};
        executorService.scheduleAtFixedRate(() -> { // 定时推送信息到客户端
            log.info("scheduleAtFixedRate-1, 执行了!"+ i[0]);
            for (SseEmitter sseEmitter : clients) {
                i[0]++;
                try {
                    Instant start = Instant.now();
                    HashMap hashMap = MapUtil.of(String.valueOf(i[0]), ":当前推送次数" + IdUtil.fastSimpleUUID());
                    // 推送信息到客户端
                    sseEmitter.send(hashMap, MediaType.APPLICATION_JSON);
                    log.info("Sent to client, took: {}", Duration.between(start, Instant.now()));
                } catch (IOException e) {
                   log.error("scheduleAtFixedRate-2, 前端关闭了连接!");
                }
            }
        }, 15, 3, TimeUnit.SECONDS);
    }
}

网站在线人数、 促销成交金额、 监控屏数据

1