基于SSE的信息推送实践

原创 Tono 拍码场

2025年04月24日 03:18

什么是 SSE

SSE（Server-Sent Events）是一种基于 HTTP 协议的服务器推送技术，允许服务器主动向客户端（如浏览器）发送实时数据更新，而无需客户端重复发起请求。该技术通过建立持久化的单向通信通道，实现了高效的实时信息传输。

SSE与其他协议对比

SSE的业务实践

在坐席业务中，电话是联系客户的主要手段。系统支持点呼和批量外呼。批量外呼可同时拨打多个电话，接通后转接坐席。但当两种呼叫方式同时使用，且坐席忙于点呼时，批量外呼接通的电话需排队等待，可能导致客户放弃等待，从而增加呼损率。

解决方案

对接语音平台话务事件，对进入队列等待转接到坐席的外呼及时通过SSE推送到前端
前端接收到队列中的话务事件，如果当前点呼拨打中（未接通），立即挂断当前电话
如果坐席在通话中，外呼看板中展示对应排队的号码，坐席可选择继续通话或挂断当前通话，等待排队中的电话转接进来

示意图

Genesys-语音平台：通话的每个状态产生话务事件，通过MQ广播出去
datalink-业务系统数据对接服务：对接MQ话务事件，通知到对应保持SSE连接的Server
home-直接对接前端的服务：维护SSE连接，推送消息到前端

消息发送流程图

一个SSE连接只能与一台Server保持连接，发送消息时需要找到对应的Server
在创建连接时，本地缓存连接对象，并将本地IP缓存到redis，key为坐席id
一个坐席只需要存一个SSE连接，在新建连接后需关闭并清除之前连接

SSE使用技巧

SSE 使用基础

NIO 支持

Tomcat 从版本 7 开始就支持 NIO（非阻塞 I/O），可以通过配置来启用 NIO
Tomcat 8 开始，默认采用 NIO（非阻塞 I/O）模式处理 HTTP 请求
Springboot 1.5.x 以上默认使用 tomcat 8，Springboot2.0 以上版本基本默认 tomcat 9

系统容量评估
SSE只有在创建、发送消息时占用tomcat工作线程，可根据情况，适当调整 tomcat 工作线程，满足 SSE 新增的消耗

SSE 连接可靠性

代码示例

后端SSE连接管理，需要注意无用连接的销毁，防止JVM内存占用过多

本地缓存SSE连接需使用线程安全的对象，如ConcurrentHashMap
前端连接断开，后端无法感知，需要设置SSE连接的超时时间
无效连接，需要调用SseEmitter.complete()销毁对象

后端需要主动断开SSE连接时，需要前端配合调用相关API关闭连接

后端调用SseEmitter.complete()后连接断开，后端服务发布等造成的连接断开，前端收到的都是error事件，此时前端无法区分是否需要断开连接还是交给浏览器重连
后端需定义事件，如CLOSE，前端收到后主动调用API关闭连接。后端再通过SseEmitter.complete()销毁连接对象

后端代码示例

@RequestMapping("/sse")public class SseController {    //本地缓存，用于存储SseEmitter    private final static ConcurrentHashMap<Long, SseEmitter> EMITTERS = new ConcurrentHashMap<>();
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)    public SseEmitter createSSE(@RequestParam Long userId) {        SseEmitter emitter = new SseEmitter(60_000L); // 60秒超时        // 连接关闭处理，移除SseEmitter        emitter.onCompletion(() -> EMITTERS.remove(userId));        emitter.onTimeout(() -> EMITTERS.remove(userId));        //本地缓存中存储SseEmitter        EMITTERS.put(userId, emitter);        // TODO redis中维护userId与当前实例IP映射        return emitter;    }
    @PostMapping(value = "/send")    public ResponseEntity<String> sendMessage(@RequestBody Message message) {        // 本地缓存获取SseEmitter        Long userId = message.getUserId();        SseEmitter sseEmitter = EMITTERS.get(userId);        try {            // 发送消息            sseEmitter.send(SseEmitter.event().name("message").data(message));        } catch (IOException e) {            // 异常时销毁对象            sseEmitter.completeWithError(e);            // 移除本地缓存            EMITTERS.remove(userId);        }        return ResponseEntity.ok("ok");    }
}

前端代码示例

<script>        // 自动创建SSE连接        window.onload = function() {            // 创建EventSource实例（替换为你的SSE端点）            const sse = new EventSource('https://domain/sse/stream?token=***');
            // 连接成功回调            sse.onopen = () => {//do business            };
            // 接收消息处理, event.type=message            sse.onmessage = (event) => {                // do business            };
            // 自定义事件处理（可选），例如以Close事件告知前端主动关闭连接
            sse.addEventListener('Close', (e) => {                // do business                sse.close();
            });
            // 错误处理            // 后端通过complete事件主动关闭连接时，如果未通过代码主动关闭连接浏览器会自动重连            sse.onerror = () => {readyState://连接已打开并准备进行通信,异常未断开连接。如后端主动推送error事件if(sse.readyState === EventSource.OPEN){					sse.close();				}                // 连接尚未打开或异常关闭，如后端的complete事件，后端服务重启，nginx reload，网络异常等...                else if(sse.readyState === EventSource.CONNECTING) {// do nothing，浏览器会自动重连，默认是3s重连一次，可修改				}            };        };    </script>