做后端这几年实时推送几乎是每个项目都绕不开的硬需求。早年做高校实验室预约系统时学生提交预约后实验员那边得刷新好几遍页面才能看到新申请后来做集成大华摄像头的监控平台设备报警要等轮询拉取才知道延迟能到好几秒。这两段经历让我对Spring Boot实时推送有了非常实在的体会它不像数据库事务那样有一个标准答案而是要根据场景选型选错了后面全是坑。我这次就用三个真实做过的案例来拆解Spring Boot实时推送的完整技术链。第一个是SSE实现的审核通知推送轻量、省资源适合服务端单向发消息第二个是WebSocket加STOMP协议做的双向聊天与工单实时互动适合你来我往的场景第三个是WebSocket对接大华SDK把设备报警、预览流地址、云台控制指令全都通过推送通道转发给前端。三套方案都由浅入深从HTTP长连接到全双工协议再到和硬件SDK联动基本涵盖了平时面试和实际项目里会碰到的所有核心考点。文章里会写清楚每个方案的原理、完整可跑的代码、参数选择依据还有那些文档里不会告诉你的坑。1. 三个实时推送方案怎么选才不踩坑1.1 从短轮询到长连接的演进逻辑很多初学者拿到实时推送需求第一反应就是前端setInterval定时器每隔两三秒发一次请求。这种方式在用户量小、实时性要求不高的后台管理系统里确实能跑但一旦并发上来问题就很明显Tomcat线程池被频繁的无效请求占满数据库也跟着遭殃——明明没什么新数据查询却一直在执行。短轮询的本质是“客户端主动问”效率低是因为绝大多数请求都拿不到新数据。后来演进出长轮询也就是服务器收到请求后不立即返回而是hold住这个连接等有新数据了再响应客户端收到后再立刻发起下一次请求。这算是半长连接减少了大量无效请求的浪费但每次响应后TCP连接都要重新建立握手开销依然存在。再往后就是真正的长连接方案SSEServer-Sent Events和WebSocket。SSE是HTTP协议上的单向通道服务器可以持续向客户端推送数据客户端只能用EventSource或fetch来接收不能往服务器发业务消息。WebSocket则是一次握手后建立TCP长连接双向收发。用大白话比喻SSE像广播电台你打开收音机听节目但不能对着收音机说话WebSocket像电话两边随时可以开口。选型时就看业务是否要求客户端往服务器频繁发指令如果只是服务端状态变更通知SSE足够如果需要聊天、指令控制这类双向交互必须上WebSocket。1.2 三个案例各自解决什么真实问题第一个案例来自高校实验室预约系统。学生提交预约申请后实验员在后台审核审核结果需要实时出现在学生端的页面上不需要学生手动刷新。这里只有服务端到客户端一个方向的数据流用SSE最合适实现成本低还能自动断线重连。第二个案例是实验室内网聊天和工单协作。老师、实验员、学生三端需要互相发消息、推送“正在输入”状态、通知工单被认领或完结。客户端不仅要收消息还要发送自己的操作所以必须用WebSocket我选的是Spring框架里最成熟的STOMP协议方案。第三个案例是集成大华SDK的实时监控系统。设备报警事件要从SDK回调线程里发出来推送到监控大屏用户点击某台摄像机的预览按钮前端要拿到实时预览流地址用户在页面上拖拽云台方向键控制指令要下发给设备。预览流地址和报警事件是单向的但云台控制是双向指令所以整体架构用WebSocket统一承载。这个案例最有价值的地方在于展示了怎么把第三方SDK的回调机制和Spring Boot的异步推送能力桥接起来。1.3 面试关心的四个技术考点如果把Spring Boot实时推送作为面试题来准备通常会被问到四个方面。第一是方案选型对比要能说清楚SSE和WebSocket的区别、各自的适用边界别一上来就说WebSocket天下第一。第二是WebSocket握手阶段怎么鉴权很多人知道用拦截器却说不好token放在哪里、怎么校验。第三是底层连接管理一个应用有成百上千个长连接session怎么存放、断线怎么清理、心跳怎么维持。第四是集群环境下推送失效怎么办单机能跑通的东西部署到多节点就出问题这通常是考察候选人有没有真实上线经验的分水岭。下面的三个案例其实就是把这四个考点逐个拆开喂给你。2. 案例一SSE实现通知消息实时推送2.1 先搞懂SseEmitter的原理和生命周期Spring Boot里实现SSE推送核心类是SseEmitter。它的工作方式和Servlet 3.1的异步响应机制挂钩Controller接收到请求后返回一个SseEmitter对象同时把当前线程释放回容器线程池请求连接保持在服务端不关闭。之后业务代码在任意线程里调用emitter.send()数据就会通过这条挂起的连接推送到客户端。理解这个机制就明白了两件事。第一Controller本身不阻塞线程所以SSE不会占用Tomcat的工作线程——这正是它比轮询省资源的关键。第二连接生命周期完全由SseEmitter控制服务端不主动调用complete()或发生异常连接就一直开着。这引出一个经典问题SseEmitter不会自己超时断开吗答案是有默认超时时间Spring Boot里默认是30秒左右但实际使用中我们必须手动设置一个较长的超时比如30分钟或1小时并且通过定时发心跳数据来保活。遇到Nginx代理时还要调整proxy_read_timeout参数否则代理层会先把空闲连接断掉。SseEmitter和客户端之间是一条单向管道所以服务端要自己维护一个“管道仓库”。我的做法是定义一个全局的ConcurrentHashMapkey用用户IDvalue是用户所有活跃连接的列表。用户多端登录时同一个userId可能对应多个SseEmitter推送时就要遍历这个列表逐个发送。2.2 接口设计与数据格式约定我做的实验室预约系统里实时推送的对象主要是两类消息审核结果通知和设备状态变更。审核结果通知是实验员操作后系统把“通过”或“驳回”的结果推给学生设备状态变更是设备被占用或释放时推送给所有关注该设备的学生和教师。接口设计遵循RESTful风格推送通道统一用GET /api/sse/subscribe。因为SSE规范规定EventSource只能发GET请求这也是SSE的一个限制。连接建立后后续数据全部走SSE的命名事件机制。我定义了几种事件类型AUDIT_RESULT、DEVICE_STATUS、PING前端注册对应的EventListener来做不同处理。消息体统一用JSON封装包含type、data、timestamp三个字段。这样前端只需要一个订阅函数根据type字段分发到不同的业务处理逻辑。对于“新预约申请到达实验员端”这个场景要注意推送的并发条件。学生提交预约后事务还没提交就推送实验员点开详情可能查到的是旧数据。我这里的经验是用Spring的TransactionalEventListener配合TransactionPhase.AFTER_COMMIT确保推送逻辑在事务提交后执行。这属于非常容易忽略的细节但线上一定会踩。2.3 完整实现代码与关键配置服务端的核心代码分三部分SseEmitter的管理器、订阅控制器、推送工具类。先看Emitter管理器我用一个单例的组件来统一管理所有连接Component public class SseConnectionManager { private final MapLong, MapString, SseEmitter userEmitters new ConcurrentHashMap(); public SseEmitter subscribe(Long userId) { SseEmitter emitter new SseEmitter(0L); // 0L表示不超时 MapString, SseEmitter emitterMap userEmitters .computeIfAbsent(userId, k - new ConcurrentHashMap()); String connectionId UUID.randomUUID().toString(); emitterMap.put(connectionId, emitter); emitter.onCompletion(() - { emitterMap.remove(connectionId); if (emitterMap.isEmpty()) { userEmitters.remove(userId); } }); emitter.onTimeout(() - { emitter.complete(); emitterMap.remove(connectionId); }); emitter.onError(e - { emitter.completeWithError(e); emitterMap.remove(connectionId); }); return emitter; } public void sendToUser(Long userId, SseEventBuilder builder) { MapString, SseEmitter emitterMap userEmitters.get(userId); if (emitterMap null) return; emitterMap.forEach((id, emitter) - { try { emitter.send(builder); } catch (IOException e) { emitter.completeWithError(e); emitterMap.remove(id); } }); } }这段代码有两个设计点需要说明。第一是SseEmitter(0L)0表示永不过期。这样连接就完全由心跳机制来控制生命周期不受Tomcat默认超时约束。第二是每个连接用UUID区分因为同一个用户可能开两个浏览器标签页得支持多路连接分别管理和清理。onCompletion、onTimeout、onError三个回调是清理时机的三保险一定要都写上否则断线用户会在Map里囤积大量死连接。然后是订阅接口和推送逻辑RestController RequestMapping(/api/sse) public class NotificationSseController { private final SseConnectionManager connectionManager; GetMapping(/subscribe) public SseEmitter subscribe(RequestParam Long userId) { SseEmitter emitter connectionManager.subscribe(userId); // 连接建立后立刻发送一条连接成功消息 try { MapString, Object data new HashMap(); data.put(message, connected); emitter.send(SseEmitter.event() .name(CONNECTED) .data(data, MediaType.APPLICATION_JSON)); } catch (IOException e) { emitter.completeWithError(e); } return emitter; } }推送工具类封装了业务方调用的入口Component public class NotificationPushService { private final SseConnectionManager connectionManager; private final ObjectMapper objectMapper; public void pushAuditResult(Long studentId, AuditResultDTO result) { MapString, Object payload new HashMap(); payload.put(orderId, result.getOrderId()); payload.put(status, result.getStatus()); payload.put(message, result.getMessage()); payload.put(timestamp, System.currentTimeMillis()); SseEmitter.SseEventBuilder builder SseEmitter.event() .name(AUDIT_RESULT) .data(payload, MediaType.APPLICATION_JSON); connectionManager.sendToUser(studentId, builder); } }前端订阅代码用原生EventSource就够了const eventSource new EventSource(/api/sse/subscribe?userId${userId}); eventSource.addEventListener(AUDIT_RESULT, (event) { const data JSON.parse(event.data); showNotification(预约${data.status PASSED ? 通过 : 被驳回}: ${data.message}); }); eventSource.addEventListener(DEVICE_STATUS, (event) { const data JSON.parse(event.data); updateDeviceStatus(data.deviceId, data.status); }); eventSource.onerror () { // 断线后浏览器会自动重连这里记录日志即可 console.warn(SSE连接断开浏览器将自动重试); };2.4 连接保活与断线清理的实战心得SSE用起来门槛低但要把海量连接维护好有几个细节必须处理到位。第一个是心跳机制。我专门写了一个定时任务每30秒对所有活跃的SseEmitter发送一个PING事件。这样做的目的不只是告诉客户端“服务端还活着”更重要的是让中间的网络代理Nginx、负载均衡器感知到连接上有流量从而避免空闲连接被回收。定时任务我用Spring的Scheduled注解实现注意固定频率要用fixedRate而不是fixedDelay因为fixedDelay要等上一次执行完才计时如果发送失败会拖慢后续心跳。第二个是消息推送失败的容忍策略。emitter.send()在被客户端断开后调用会抛出IOException这是正常的。真正要注意的是不要在业务主流程里同步推送。比如审核结果保存成功后如果推送恰好失败不能因此回滚数据库事务。我的做法是推送逻辑不管成功失败都不影响主流程失败只记录日志客户端重连后可以通过查询接口补齐遗漏的消息。换句话说实时推送是锦上添花最终一致性还得靠普通接口兜底。第三个是内存视角的考量。一个SseEmitter占用几十KB内存一万个连接就是几百MB所以每个User的EmitterMap必须有上限控制。我在代码里加了MAX_CONNECTIONS_PER_USER 5的检查超过就淘汰最早的连接。这样可以防止用户一直刷新页面导致连接泄漏。3. 案例二WebSocket STOMP实现双向实时互动3.1 为什么不用原生WebSocket而用STOMP原生WebSocket的API非常简单服务端只需要一个WebSocketHandler重写afterConnectionEstablished、handleTextMessage这些方法。但业务稍微复杂一点问题就来了消息怎么路由怎么区分发给指定用户还是全体广播连接要不要按业务类型分别处理STOMP的出现就是为了解决这些协议层之上的语义问题。它是构建在WebSocket之上的一层简单消息协议定义了CONNECT、SUBSCRIBE、SEND这些帧类型。你可以把STOMP理解成HTTP之于TCP——TCP提供了双向字节流HTTP定义了请求和响应的语义WebSocket提供了双向消息通道STOMP则在上面定义了谁订阅什么主题、谁能向哪个地址发消息。Spring Boot对STOMP的支持很成熟用MessageMapping注解声明消息处理器用SendTo或SendToUser指定返回数据发到哪个主题代码写起来很像Controller。选择STOMP带来的另一个好处是Spring Security的天然整合。鉴权用户在WebSocket建立后会被自动放入Principal对象SendToUser注解利用这个Principal定向分发不用自己解析用户身份。3.2 三层结构配置类、消息控制器、前端订阅先上依赖Maven项目在pom里引入dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency接下来说配置。WebSocket的STOMP配置核心是注册端点、设置代理和目标前缀Configuration EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { Override public void registerStompEndpoints(StompEndpointRegistry registry) { registry.addEndpoint(/ws-chat) .setAllowedOriginPatterns(*) .addInterceptors(new AuthHandshakeInterceptor()) .withSockJS(); } Override public void configureMessageBroker(MessageBrokerRegistry registry) { registry.enableSimpleBroker(/topic, /queue); registry.setApplicationDestinationPrefixes(/app); registry.setUserDestinationPrefix(/user); } }这里有两个前缀很容易搞混。setApplicationDestinationPrefixes(/app)是客户端发送消息时用的前缀客户端往/app/chat/send发送的消息会路由到MessageMapping(/chat/send)方法。enableSimpleBroker(/topic, /queue)是服务端转发消息主题的前缀/topic开头的是广播主题所有人都能订阅/queue开头的是点对点队列只有指定用户能收到。聊天的消息控制器是这样写的Controller public class ChatMessageController { MessageMapping(/chat/send) SendTo(/topic/room/{roomId}) public ChatMessage broadcast(ChatMessage message, Principal principal) { message.setFrom(principal.getName()); message.setTimestamp(System.currentTimeMillis()); return message; } MessageMapping(/chat/private) public void sendPrivate(PrivateMessage message, Principal principal) { message.setFrom(principal.getName()); messagingTemplate.convertAndSendToUser( message.getTo(), /queue/private, message); } }注意SendTo(/topic/room/{roomId})这里的占位符是Spring Messaging框架根据消息头自动解析的。如果用的是普通SendTo(/topic/room)就没有这个能力。我实际开发时更常用SimpMessagingTemplate来手动指定动态主题灵活度更高。前端订阅代码用STOMP客户端创建连接、订阅主题、发送消息三步const socket new SockJS(/ws-chat); const stompClient Stomp.over(socket); stompClient.connect({}, (frame) { stompClient.subscribe(/topic/room/${roomId}, (message) { const chat JSON.parse(message.body); renderMessage(chat); }); stompClient.subscribe(/user/queue/private, (message) { const privateMsg JSON.parse(message.body); showPrivateMessage(privateMsg); }); }); function sendMessage() { stompClient.send(/app/chat/send, {}, JSON.stringify({ roomId: currentRoomId, content: input.value })); }3.3 握手鉴权把Spring Security的令牌检查前置WebSocket端点如果裸奔任何人都可以连接上来订阅敏感主题这跟actuator端点暴露在公网一样危险。我在配置类里注册了AuthHandshakeInterceptor在TCP握手阶段拦截请求校验token。常见的做法是客户端在URL写成/ws-chat?tokenxxx拦截器解析query里的token并校验public class AuthHandshakeInterceptor implements HandshakeInterceptor { Override public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, MapString, Object attributes) { String token getTokenFromQuery(request); if (StringUtils.hasText(token)) { UserInfo userInfo JwtTokenUtil.parseToken(token); if (userInfo ! null) { attributes.put(userId, userInfo.getUserId()); attributes.put(username, userInfo.getUsername()); return true; } } response.setStatusCode(HttpStatus.UNAUTHORIZED); return false; } }把解析出的用户信息放入attributes之后在WebSocketHandler里能通过session.getAttributes()取出来。这样就不需要在业务处理器里二次解析token了。有一点必须提醒浏览器端SockJS握手时自定义header会被限制所以token只能放URL的query或者通过connect方法的header参数。我推荐的方案是query传token虽然URL里带token不太好看但WebSocket的query本来就是这么用的比header兼容性好。3.4 集群环境的广播失效与Redis适配单机部署时simple broker是运行在JVM内存里的完全没问题。一旦部署多个实例问题立刻暴露用户A连在机器1上用户B连在机器2上A发消息给B消息只会经过A所在机器1的broker机器2上的B收不到。这就是典型的“广播不出城堡”。生产环境我建议在Spring Boot里用RabbitMQ或ActiveMQ作为外部broker替代内存版simple broker。配置改成这样registry.enableStompBrokerRelay(/topic, /queue) .setRelayHost(rabbitmq.internal) .setRelayPort(61613) .setClientLogin(guest) .setClientPasscode(guest);这样所有实例都订阅同一个外部brokerA发的消息先进RabbitMQ再由RabbitMQ推给所有实例最终到达B。如果团队不想引入消息中间件也有个折中方案应用内自己实现基于Redis发布订阅的broker但问题很多可靠性、消息顺序、持久化都要自己写我不建议在项目中自己造这个轮子。3.5 点对点推送的坑SendToUser为什么偶尔丢消息SendToUser的实现原理是框架在内部把用户目标转成一个特殊的主题/user/{username}/queue/xx并且只有当前会话能订阅。但有个坑是如果用户开了两个标签页两个会话都订阅了同一个/user/queue/xx默认情况下只有其中一个会话能收到消息另一个不会。原因是Spring的UserDestinationMessageHandler会把发往用户目标的消息解析为发往一个特定sessionId的主题。要调整这个行为需要自己指定sessionId。我踩过这个坑之后在SimpMessagingTemplate.convertAndSendToUser方法里传入了用户的sessionId作为第三个参数messagingTemplate.convertAndSendToUser( userId.toString(), /queue/private, message, createHeaders(sessionId));这样每条消息都能精确到达指定的会话。4. 案例三WebSocket 大华SDK实现设备实时监控推送4.1 项目背景集成预览、回放、云台控制这个项目的需求来自一个园区安防平台。平台需要接入园区内的大华网络摄像机在Web端实现三块核心功能实时预览监控画面、历史录像回放、云台方向控制。同时设备产生的报警事件移动侦测、遮挡报警、信号丢失要实时推送到监控中心的大屏上。大华SDK提供了完整的设备接入能力但它是一个基于JNAJava Native Access的SDK底层通过调用C语言动态库来完成登录设备、获取流、订阅报警等操作。SDK回调事件是以独立线程的方式触发的SDK内部维护了一套回调机制当设备有报警时相应的回调函数在SDK的线程池里被调用。问题在于这些回调线程跟Spring Boot应用自身的线程模型完全隔离我们必须在回调里拿到数据后交给Spring管理的消息链路再推送到前端WebSocket会话。4.2 整体架构SDK回调怎么接到WebSocket通道最忌讳的做法是直接在SDK回调函数里写WebSocket的send逻辑。原因有三个第一SDK回调线程通常是高优先级的原生线程如果在这里做耗时操作会阻塞后续回调的执行导致设备事件丢失第二回调函数不归Spring容器管理无法直接注入业务Service第三回调线程没有Spring的上下文异常处理也不好做。我的方案是在中间加一层事件桥接。SDK回调线程只做一件事构造一个轻量级的事件对象设备序列号、通道号、事件类型、时间戳然后把它发布到Spring的事件机制中。Spring的Async监听器会异步接收这个事件再调用WebSocket推送服务把数据转换成JSON后发给对应的订阅会话。这样做的价值在于把SDK线程和业务线程彻底解耦。SDK回调只花不到1毫秒就返回不会阻塞设备端真正耗时的推送逻辑异步执行还有完整的Spring事务和异常体系。4.3 核心代码报警转发与预览流地址推送先定义一个设备事件对象public class DeviceEvent { private String deviceSerial; private Integer channel; private Integer eventType; private Long timestamp; private MapString, Object extraData; }然后是SDK回调到Spring事件的桥接Component public class DahuaEventDispatcher { private final ApplicationEventPublisher eventPublisher; public void dispatchAlarm(String deviceSerial, int channel, int eventType, MapString, Object extraData) { DeviceEvent event new DeviceEvent(); event.setDeviceSerial(deviceSerial); event.setChannel(channel); event.setEventType(eventType); event.setTimestamp(System.currentTimeMillis()); event.setExtraData(extraData); // 异步发布避免阻塞SDK回调线程 eventPublisher.publishEvent(event); } }需要说明的是这里dispatchAlarm方法是在大华SDK的报警回调里直接调用的所以它在SDK的原生线程上执行。我给它标注了极短的时间复杂度保证只负责创建对象和发布事件。Spring事件监听器异步接收并推送Component public class AlarmPushListener { private final SimpMessagingTemplate messagingTemplate; Async EventListener public void onDeviceEvent(DeviceEvent event) { MapString, Object payload new HashMap(); payload.put(deviceSerial, event.getDeviceSerial()); payload.put(channel, event.getChannel()); payload.put(eventType, event.getEventType()); payload.put(timestamp, event.getTimestamp()); // 推送到所有订阅设备报警主题的终端 messagingTemplate.convertAndSend(/topic/alarm/ event.getDeviceSerial(), payload); } }这里我用的是前面案例二搭好的WebSocket STOMP通道。监控大屏前端只需要在启动时订阅对应设备的报警主题报警事件就能实时上屏。预览流地址的推送逻辑也是类似的思路。用户点击“预览”按钮时请求到达Spring的Controller应用调用大华SDK的实时预览接口拿到RTSP取流URL或者HCVWSS协议地址然后通过WebSocket推送给当前会话。云台控制则反过来前端通过STOMP的send指令把方向和步长发给服务端服务端调用大华SDK的云台控制接口下发给设备。这正好用上了WebSocket双向通信的能力单向SSE在这里是彻底的死路。4.4 大华SDK集成的三个典型坑点这个案例在线上的运维过程中我踩过大华SDK相关的三个坑都很有代表性。第一个是SDK初始化全局唯一。大华SDK的NET_DVR_Init()方法必须在应用启动时调用一次并且全局只能初始化一次。如果多个线程并发调用初始化会导致SDK内部状态混乱轻则设备登录失败重则进程崩溃。我的做法是用一个PostConstruct方法在应用启动阶段显式初始化同时通过一个静态标志位防止重复初始化。第二个是设备登录句柄需要Careful管理。SDK登录后返回一个用户句柄后续所有操作预览、云台控制、报警布防都依赖这个句柄。句柄用完后必须调用登出接口释放否则句柄数量过多会导致SDK资源耗尽。但句柄释放的时机很讲究报警布防的回调还在进行中句柄被释放就会引发回调线程访问野指针导致JVM崩溃。我的经验是给每个句柄加引用计数所有业务操作结束后才真正释放。第三个是JNA内存释放。大华SDK的Java接口通过JNA调用的方式传递结构体时如果涉及指针类型的参数必须在调用完成后手动释放内存。我遇到过一个情况长时间跑下来内存不断增长用MAT分析堆没发现问题后来发现是JNA直接从native堆分配的DirectByteBuffer没有被释放。这个问题的排查非常隐蔽解决方式是在SDK调用工具类里显式调用Pointer.release()或者用try-finally保证释放逻辑一定执行。4.5 推送通道的安全加固从actuator暴露说起的鉴权必要性提到actuator很多人第一时间想到的是生产环境忘了关actuator端点结果内部信息被扫出来。这个思路放在WebSocket推送通道上完全同理。一个只校验了连接来源、没校验用户身份的WebSocket端点就跟对外开放的actuator一样危险——任何人都能订阅/topic/alarm/*主题监控信息直接泄露任何人都能往云台控制主题发指令设备就会被非法操控。我给这个项目加的保护分两层。第一层是握手阶段鉴权所有WebSocket连接必须带有效的JWT这跟case二里的拦截器是一致的。第二层是主题级别授权在ChannelInterceptor里对进入的SUBSCRIBE帧做权限校验只允许用户订阅自己权限范围内的主题。代码要点是拦截MessageType.SUBSCRIBE类型的消息解析目标主题再结合当前Principal判断权限Override public Message? preSend(Message? message, MessageChannel channel) { StompHeaderAccessor accessor StompHeaderAccessor.wrap(message); if (MessageType.SUBSCRIBE.equals(accessor.getCommandType())) { String destination accessor.getDestination(); Principal principal accessor.getUser(); if (!permissionService.canSubscribe(principal, destination)) { throw new AccessDeniedException(无权订阅此主题); } } return message; }5. 常见问题与性能调优实录5.1 实时推送高频问题速查表我在几个项目里积累了一份问题清单基本上新人在实时推送上踩的坑都集中在这张表里现象根因解决思路SSE连接几十秒后自动断开默认超时时间或代理层空闲超时设置SseEmitter无超时加30秒心跳WebSocket能握手但订阅后收不到消息端点前缀或订阅主题路径不一致检查/app、/topic、/user三个前缀的分工广播消息部分用户收不到集群部署但没配StompBrokerRelay引入RabbitMQ或更换外部broker用户多端登录时定向消息只发到一端SendToUser默认发到单一会话用SimpMessagingTemplate指定sessionId消息推送偶尔乱序业务在事务提交前推送使用TransactionalEventListener(AFTER_COMMIT)Tomcat端口溢出长连接没释放确保onCompletion/onTimeout/onError三清理齐全服务端重启后客户端不自动恢复需要客户端重连机制SockJS自带重连SSE用EventSource的onerror处理这张表在技术分享会上反复用过大家可以对照排查。5.2 用Actuator和Micrometer给推送连接装上仪表盘实时推送功能上线后运维最关心的问题就是现在有多少个活跃连接消息推送速率是多少失败率有没有升高这部分我用Actuator加Micrometer做了运行指标监控。Metal监控的核心是把推送通道的实时状态暴露成Metrics。我做的是在SseConnectionManager里维护一个activeConnectionCount的AtomicLong字段用Micrometer的Gauge注册到MeterRegistryComponent public class PushMetrics { public PushMetrics(MeterRegistry registry, SseConnectionManager connectionManager) { Gauge.builder(push.sse.active.connections, connectionManager, SseConnectionManager::getActiveConnectionCount) .description(活跃的SSE连接数) .register(registry); } }然后在application.yaml里开启actuator的metrics端点配合Prometheus采集management: endpoints: web: exposure: include: health,info,metrics,prometheus metrics: export: prometheus: enabled: true这样Grafana上就能实时看到活跃连接数、WebSocket会话数、每分钟推送消息数。我还加了基于连接数的告警规则单节点活跃连接超过5000就预警因为这说明容量接近瓶颈该考虑扩容了。5.3 线程池隔离与前端重连策略推送功能对线程资源的消耗和普通HTTP请求完全不同最怕的是被其他耗时接口拖垮。我在项目中专门为推送逻辑配置了独立的异步线程池核心线程数、最大线程数、队列容量都跟业务线程池分开。这样即使某个业务的慢SQL把默认线程池撑爆了推送线程池依然能正常工作保证用户能实时收到报警。前端重连策略也是实时推送体验的重要一环。SSE里EventSource自带自动重连但重连间隔是浏览器默认的大约3秒服务器繁忙时会造成重连风暴。我建议服务端在发送错误时主动返回一个特殊事件RECONNECT_DELAY客户端收到后手动设置重连延迟做指数退避。WebSocket的客户端就要自己做重连逻辑了我的做法是在onclose回调里用setTimeout重连间隔从1秒开始每次翻倍最多30秒封顶避免服务端故障恢复瞬间所有客户端同时砸过来。另外一个小技巧服务端推送消息前尽量先做一次session.isOpen()检查避免往已经关闭的会话发数据。WebSocket会因为心跳失败、网络闪断等各种原因在服务端还没感知时就失效了主动检查能减少大量无效尝试。实时推送做了这么多年我的体会是它从来都不是一个单一注解或单一框架能搞定的事情。SSE也好、WebSocket也好甚至后面再接上RabbitMQ、Redis、大华SDK本质都是在解决同一个问题让数据在正确的时间以正确的姿态到达正确的终端。技术选型只是第一步连接管理、鉴权隔离、断线重连、集群扩展、监控告警每一项都需要在实际项目里真刀真枪地磨一遍。希望这三个案例能给正在做或准备做实时推送的朋友提供一个完整的参考少走几步弯路。
Spring Boot实时推送实战：SSE、WebSocket与STOMP选型指南
做后端这几年实时推送几乎是每个项目都绕不开的硬需求。早年做高校实验室预约系统时学生提交预约后实验员那边得刷新好几遍页面才能看到新申请后来做集成大华摄像头的监控平台设备报警要等轮询拉取才知道延迟能到好几秒。这两段经历让我对Spring Boot实时推送有了非常实在的体会它不像数据库事务那样有一个标准答案而是要根据场景选型选错了后面全是坑。我这次就用三个真实做过的案例来拆解Spring Boot实时推送的完整技术链。第一个是SSE实现的审核通知推送轻量、省资源适合服务端单向发消息第二个是WebSocket加STOMP协议做的双向聊天与工单实时互动适合你来我往的场景第三个是WebSocket对接大华SDK把设备报警、预览流地址、云台控制指令全都通过推送通道转发给前端。三套方案都由浅入深从HTTP长连接到全双工协议再到和硬件SDK联动基本涵盖了平时面试和实际项目里会碰到的所有核心考点。文章里会写清楚每个方案的原理、完整可跑的代码、参数选择依据还有那些文档里不会告诉你的坑。1. 三个实时推送方案怎么选才不踩坑1.1 从短轮询到长连接的演进逻辑很多初学者拿到实时推送需求第一反应就是前端setInterval定时器每隔两三秒发一次请求。这种方式在用户量小、实时性要求不高的后台管理系统里确实能跑但一旦并发上来问题就很明显Tomcat线程池被频繁的无效请求占满数据库也跟着遭殃——明明没什么新数据查询却一直在执行。短轮询的本质是“客户端主动问”效率低是因为绝大多数请求都拿不到新数据。后来演进出长轮询也就是服务器收到请求后不立即返回而是hold住这个连接等有新数据了再响应客户端收到后再立刻发起下一次请求。这算是半长连接减少了大量无效请求的浪费但每次响应后TCP连接都要重新建立握手开销依然存在。再往后就是真正的长连接方案SSEServer-Sent Events和WebSocket。SSE是HTTP协议上的单向通道服务器可以持续向客户端推送数据客户端只能用EventSource或fetch来接收不能往服务器发业务消息。WebSocket则是一次握手后建立TCP长连接双向收发。用大白话比喻SSE像广播电台你打开收音机听节目但不能对着收音机说话WebSocket像电话两边随时可以开口。选型时就看业务是否要求客户端往服务器频繁发指令如果只是服务端状态变更通知SSE足够如果需要聊天、指令控制这类双向交互必须上WebSocket。1.2 三个案例各自解决什么真实问题第一个案例来自高校实验室预约系统。学生提交预约申请后实验员在后台审核审核结果需要实时出现在学生端的页面上不需要学生手动刷新。这里只有服务端到客户端一个方向的数据流用SSE最合适实现成本低还能自动断线重连。第二个案例是实验室内网聊天和工单协作。老师、实验员、学生三端需要互相发消息、推送“正在输入”状态、通知工单被认领或完结。客户端不仅要收消息还要发送自己的操作所以必须用WebSocket我选的是Spring框架里最成熟的STOMP协议方案。第三个案例是集成大华SDK的实时监控系统。设备报警事件要从SDK回调线程里发出来推送到监控大屏用户点击某台摄像机的预览按钮前端要拿到实时预览流地址用户在页面上拖拽云台方向键控制指令要下发给设备。预览流地址和报警事件是单向的但云台控制是双向指令所以整体架构用WebSocket统一承载。这个案例最有价值的地方在于展示了怎么把第三方SDK的回调机制和Spring Boot的异步推送能力桥接起来。1.3 面试关心的四个技术考点如果把Spring Boot实时推送作为面试题来准备通常会被问到四个方面。第一是方案选型对比要能说清楚SSE和WebSocket的区别、各自的适用边界别一上来就说WebSocket天下第一。第二是WebSocket握手阶段怎么鉴权很多人知道用拦截器却说不好token放在哪里、怎么校验。第三是底层连接管理一个应用有成百上千个长连接session怎么存放、断线怎么清理、心跳怎么维持。第四是集群环境下推送失效怎么办单机能跑通的东西部署到多节点就出问题这通常是考察候选人有没有真实上线经验的分水岭。下面的三个案例其实就是把这四个考点逐个拆开喂给你。2. 案例一SSE实现通知消息实时推送2.1 先搞懂SseEmitter的原理和生命周期Spring Boot里实现SSE推送核心类是SseEmitter。它的工作方式和Servlet 3.1的异步响应机制挂钩Controller接收到请求后返回一个SseEmitter对象同时把当前线程释放回容器线程池请求连接保持在服务端不关闭。之后业务代码在任意线程里调用emitter.send()数据就会通过这条挂起的连接推送到客户端。理解这个机制就明白了两件事。第一Controller本身不阻塞线程所以SSE不会占用Tomcat的工作线程——这正是它比轮询省资源的关键。第二连接生命周期完全由SseEmitter控制服务端不主动调用complete()或发生异常连接就一直开着。这引出一个经典问题SseEmitter不会自己超时断开吗答案是有默认超时时间Spring Boot里默认是30秒左右但实际使用中我们必须手动设置一个较长的超时比如30分钟或1小时并且通过定时发心跳数据来保活。遇到Nginx代理时还要调整proxy_read_timeout参数否则代理层会先把空闲连接断掉。SseEmitter和客户端之间是一条单向管道所以服务端要自己维护一个“管道仓库”。我的做法是定义一个全局的ConcurrentHashMapkey用用户IDvalue是用户所有活跃连接的列表。用户多端登录时同一个userId可能对应多个SseEmitter推送时就要遍历这个列表逐个发送。2.2 接口设计与数据格式约定我做的实验室预约系统里实时推送的对象主要是两类消息审核结果通知和设备状态变更。审核结果通知是实验员操作后系统把“通过”或“驳回”的结果推给学生设备状态变更是设备被占用或释放时推送给所有关注该设备的学生和教师。接口设计遵循RESTful风格推送通道统一用GET /api/sse/subscribe。因为SSE规范规定EventSource只能发GET请求这也是SSE的一个限制。连接建立后后续数据全部走SSE的命名事件机制。我定义了几种事件类型AUDIT_RESULT、DEVICE_STATUS、PING前端注册对应的EventListener来做不同处理。消息体统一用JSON封装包含type、data、timestamp三个字段。这样前端只需要一个订阅函数根据type字段分发到不同的业务处理逻辑。对于“新预约申请到达实验员端”这个场景要注意推送的并发条件。学生提交预约后事务还没提交就推送实验员点开详情可能查到的是旧数据。我这里的经验是用Spring的TransactionalEventListener配合TransactionPhase.AFTER_COMMIT确保推送逻辑在事务提交后执行。这属于非常容易忽略的细节但线上一定会踩。2.3 完整实现代码与关键配置服务端的核心代码分三部分SseEmitter的管理器、订阅控制器、推送工具类。先看Emitter管理器我用一个单例的组件来统一管理所有连接Component public class SseConnectionManager { private final MapLong, MapString, SseEmitter userEmitters new ConcurrentHashMap(); public SseEmitter subscribe(Long userId) { SseEmitter emitter new SseEmitter(0L); // 0L表示不超时 MapString, SseEmitter emitterMap userEmitters .computeIfAbsent(userId, k - new ConcurrentHashMap()); String connectionId UUID.randomUUID().toString(); emitterMap.put(connectionId, emitter); emitter.onCompletion(() - { emitterMap.remove(connectionId); if (emitterMap.isEmpty()) { userEmitters.remove(userId); } }); emitter.onTimeout(() - { emitter.complete(); emitterMap.remove(connectionId); }); emitter.onError(e - { emitter.completeWithError(e); emitterMap.remove(connectionId); }); return emitter; } public void sendToUser(Long userId, SseEventBuilder builder) { MapString, SseEmitter emitterMap userEmitters.get(userId); if (emitterMap null) return; emitterMap.forEach((id, emitter) - { try { emitter.send(builder); } catch (IOException e) { emitter.completeWithError(e); emitterMap.remove(id); } }); } }这段代码有两个设计点需要说明。第一是SseEmitter(0L)0表示永不过期。这样连接就完全由心跳机制来控制生命周期不受Tomcat默认超时约束。第二是每个连接用UUID区分因为同一个用户可能开两个浏览器标签页得支持多路连接分别管理和清理。onCompletion、onTimeout、onError三个回调是清理时机的三保险一定要都写上否则断线用户会在Map里囤积大量死连接。然后是订阅接口和推送逻辑RestController RequestMapping(/api/sse) public class NotificationSseController { private final SseConnectionManager connectionManager; GetMapping(/subscribe) public SseEmitter subscribe(RequestParam Long userId) { SseEmitter emitter connectionManager.subscribe(userId); // 连接建立后立刻发送一条连接成功消息 try { MapString, Object data new HashMap(); data.put(message, connected); emitter.send(SseEmitter.event() .name(CONNECTED) .data(data, MediaType.APPLICATION_JSON)); } catch (IOException e) { emitter.completeWithError(e); } return emitter; } }推送工具类封装了业务方调用的入口Component public class NotificationPushService { private final SseConnectionManager connectionManager; private final ObjectMapper objectMapper; public void pushAuditResult(Long studentId, AuditResultDTO result) { MapString, Object payload new HashMap(); payload.put(orderId, result.getOrderId()); payload.put(status, result.getStatus()); payload.put(message, result.getMessage()); payload.put(timestamp, System.currentTimeMillis()); SseEmitter.SseEventBuilder builder SseEmitter.event() .name(AUDIT_RESULT) .data(payload, MediaType.APPLICATION_JSON); connectionManager.sendToUser(studentId, builder); } }前端订阅代码用原生EventSource就够了const eventSource new EventSource(/api/sse/subscribe?userId${userId}); eventSource.addEventListener(AUDIT_RESULT, (event) { const data JSON.parse(event.data); showNotification(预约${data.status PASSED ? 通过 : 被驳回}: ${data.message}); }); eventSource.addEventListener(DEVICE_STATUS, (event) { const data JSON.parse(event.data); updateDeviceStatus(data.deviceId, data.status); }); eventSource.onerror () { // 断线后浏览器会自动重连这里记录日志即可 console.warn(SSE连接断开浏览器将自动重试); };2.4 连接保活与断线清理的实战心得SSE用起来门槛低但要把海量连接维护好有几个细节必须处理到位。第一个是心跳机制。我专门写了一个定时任务每30秒对所有活跃的SseEmitter发送一个PING事件。这样做的目的不只是告诉客户端“服务端还活着”更重要的是让中间的网络代理Nginx、负载均衡器感知到连接上有流量从而避免空闲连接被回收。定时任务我用Spring的Scheduled注解实现注意固定频率要用fixedRate而不是fixedDelay因为fixedDelay要等上一次执行完才计时如果发送失败会拖慢后续心跳。第二个是消息推送失败的容忍策略。emitter.send()在被客户端断开后调用会抛出IOException这是正常的。真正要注意的是不要在业务主流程里同步推送。比如审核结果保存成功后如果推送恰好失败不能因此回滚数据库事务。我的做法是推送逻辑不管成功失败都不影响主流程失败只记录日志客户端重连后可以通过查询接口补齐遗漏的消息。换句话说实时推送是锦上添花最终一致性还得靠普通接口兜底。第三个是内存视角的考量。一个SseEmitter占用几十KB内存一万个连接就是几百MB所以每个User的EmitterMap必须有上限控制。我在代码里加了MAX_CONNECTIONS_PER_USER 5的检查超过就淘汰最早的连接。这样可以防止用户一直刷新页面导致连接泄漏。3. 案例二WebSocket STOMP实现双向实时互动3.1 为什么不用原生WebSocket而用STOMP原生WebSocket的API非常简单服务端只需要一个WebSocketHandler重写afterConnectionEstablished、handleTextMessage这些方法。但业务稍微复杂一点问题就来了消息怎么路由怎么区分发给指定用户还是全体广播连接要不要按业务类型分别处理STOMP的出现就是为了解决这些协议层之上的语义问题。它是构建在WebSocket之上的一层简单消息协议定义了CONNECT、SUBSCRIBE、SEND这些帧类型。你可以把STOMP理解成HTTP之于TCP——TCP提供了双向字节流HTTP定义了请求和响应的语义WebSocket提供了双向消息通道STOMP则在上面定义了谁订阅什么主题、谁能向哪个地址发消息。Spring Boot对STOMP的支持很成熟用MessageMapping注解声明消息处理器用SendTo或SendToUser指定返回数据发到哪个主题代码写起来很像Controller。选择STOMP带来的另一个好处是Spring Security的天然整合。鉴权用户在WebSocket建立后会被自动放入Principal对象SendToUser注解利用这个Principal定向分发不用自己解析用户身份。3.2 三层结构配置类、消息控制器、前端订阅先上依赖Maven项目在pom里引入dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency接下来说配置。WebSocket的STOMP配置核心是注册端点、设置代理和目标前缀Configuration EnableWebSocketMessageBroker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { Override public void registerStompEndpoints(StompEndpointRegistry registry) { registry.addEndpoint(/ws-chat) .setAllowedOriginPatterns(*) .addInterceptors(new AuthHandshakeInterceptor()) .withSockJS(); } Override public void configureMessageBroker(MessageBrokerRegistry registry) { registry.enableSimpleBroker(/topic, /queue); registry.setApplicationDestinationPrefixes(/app); registry.setUserDestinationPrefix(/user); } }这里有两个前缀很容易搞混。setApplicationDestinationPrefixes(/app)是客户端发送消息时用的前缀客户端往/app/chat/send发送的消息会路由到MessageMapping(/chat/send)方法。enableSimpleBroker(/topic, /queue)是服务端转发消息主题的前缀/topic开头的是广播主题所有人都能订阅/queue开头的是点对点队列只有指定用户能收到。聊天的消息控制器是这样写的Controller public class ChatMessageController { MessageMapping(/chat/send) SendTo(/topic/room/{roomId}) public ChatMessage broadcast(ChatMessage message, Principal principal) { message.setFrom(principal.getName()); message.setTimestamp(System.currentTimeMillis()); return message; } MessageMapping(/chat/private) public void sendPrivate(PrivateMessage message, Principal principal) { message.setFrom(principal.getName()); messagingTemplate.convertAndSendToUser( message.getTo(), /queue/private, message); } }注意SendTo(/topic/room/{roomId})这里的占位符是Spring Messaging框架根据消息头自动解析的。如果用的是普通SendTo(/topic/room)就没有这个能力。我实际开发时更常用SimpMessagingTemplate来手动指定动态主题灵活度更高。前端订阅代码用STOMP客户端创建连接、订阅主题、发送消息三步const socket new SockJS(/ws-chat); const stompClient Stomp.over(socket); stompClient.connect({}, (frame) { stompClient.subscribe(/topic/room/${roomId}, (message) { const chat JSON.parse(message.body); renderMessage(chat); }); stompClient.subscribe(/user/queue/private, (message) { const privateMsg JSON.parse(message.body); showPrivateMessage(privateMsg); }); }); function sendMessage() { stompClient.send(/app/chat/send, {}, JSON.stringify({ roomId: currentRoomId, content: input.value })); }3.3 握手鉴权把Spring Security的令牌检查前置WebSocket端点如果裸奔任何人都可以连接上来订阅敏感主题这跟actuator端点暴露在公网一样危险。我在配置类里注册了AuthHandshakeInterceptor在TCP握手阶段拦截请求校验token。常见的做法是客户端在URL写成/ws-chat?tokenxxx拦截器解析query里的token并校验public class AuthHandshakeInterceptor implements HandshakeInterceptor { Override public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, MapString, Object attributes) { String token getTokenFromQuery(request); if (StringUtils.hasText(token)) { UserInfo userInfo JwtTokenUtil.parseToken(token); if (userInfo ! null) { attributes.put(userId, userInfo.getUserId()); attributes.put(username, userInfo.getUsername()); return true; } } response.setStatusCode(HttpStatus.UNAUTHORIZED); return false; } }把解析出的用户信息放入attributes之后在WebSocketHandler里能通过session.getAttributes()取出来。这样就不需要在业务处理器里二次解析token了。有一点必须提醒浏览器端SockJS握手时自定义header会被限制所以token只能放URL的query或者通过connect方法的header参数。我推荐的方案是query传token虽然URL里带token不太好看但WebSocket的query本来就是这么用的比header兼容性好。3.4 集群环境的广播失效与Redis适配单机部署时simple broker是运行在JVM内存里的完全没问题。一旦部署多个实例问题立刻暴露用户A连在机器1上用户B连在机器2上A发消息给B消息只会经过A所在机器1的broker机器2上的B收不到。这就是典型的“广播不出城堡”。生产环境我建议在Spring Boot里用RabbitMQ或ActiveMQ作为外部broker替代内存版simple broker。配置改成这样registry.enableStompBrokerRelay(/topic, /queue) .setRelayHost(rabbitmq.internal) .setRelayPort(61613) .setClientLogin(guest) .setClientPasscode(guest);这样所有实例都订阅同一个外部brokerA发的消息先进RabbitMQ再由RabbitMQ推给所有实例最终到达B。如果团队不想引入消息中间件也有个折中方案应用内自己实现基于Redis发布订阅的broker但问题很多可靠性、消息顺序、持久化都要自己写我不建议在项目中自己造这个轮子。3.5 点对点推送的坑SendToUser为什么偶尔丢消息SendToUser的实现原理是框架在内部把用户目标转成一个特殊的主题/user/{username}/queue/xx并且只有当前会话能订阅。但有个坑是如果用户开了两个标签页两个会话都订阅了同一个/user/queue/xx默认情况下只有其中一个会话能收到消息另一个不会。原因是Spring的UserDestinationMessageHandler会把发往用户目标的消息解析为发往一个特定sessionId的主题。要调整这个行为需要自己指定sessionId。我踩过这个坑之后在SimpMessagingTemplate.convertAndSendToUser方法里传入了用户的sessionId作为第三个参数messagingTemplate.convertAndSendToUser( userId.toString(), /queue/private, message, createHeaders(sessionId));这样每条消息都能精确到达指定的会话。4. 案例三WebSocket 大华SDK实现设备实时监控推送4.1 项目背景集成预览、回放、云台控制这个项目的需求来自一个园区安防平台。平台需要接入园区内的大华网络摄像机在Web端实现三块核心功能实时预览监控画面、历史录像回放、云台方向控制。同时设备产生的报警事件移动侦测、遮挡报警、信号丢失要实时推送到监控中心的大屏上。大华SDK提供了完整的设备接入能力但它是一个基于JNAJava Native Access的SDK底层通过调用C语言动态库来完成登录设备、获取流、订阅报警等操作。SDK回调事件是以独立线程的方式触发的SDK内部维护了一套回调机制当设备有报警时相应的回调函数在SDK的线程池里被调用。问题在于这些回调线程跟Spring Boot应用自身的线程模型完全隔离我们必须在回调里拿到数据后交给Spring管理的消息链路再推送到前端WebSocket会话。4.2 整体架构SDK回调怎么接到WebSocket通道最忌讳的做法是直接在SDK回调函数里写WebSocket的send逻辑。原因有三个第一SDK回调线程通常是高优先级的原生线程如果在这里做耗时操作会阻塞后续回调的执行导致设备事件丢失第二回调函数不归Spring容器管理无法直接注入业务Service第三回调线程没有Spring的上下文异常处理也不好做。我的方案是在中间加一层事件桥接。SDK回调线程只做一件事构造一个轻量级的事件对象设备序列号、通道号、事件类型、时间戳然后把它发布到Spring的事件机制中。Spring的Async监听器会异步接收这个事件再调用WebSocket推送服务把数据转换成JSON后发给对应的订阅会话。这样做的价值在于把SDK线程和业务线程彻底解耦。SDK回调只花不到1毫秒就返回不会阻塞设备端真正耗时的推送逻辑异步执行还有完整的Spring事务和异常体系。4.3 核心代码报警转发与预览流地址推送先定义一个设备事件对象public class DeviceEvent { private String deviceSerial; private Integer channel; private Integer eventType; private Long timestamp; private MapString, Object extraData; }然后是SDK回调到Spring事件的桥接Component public class DahuaEventDispatcher { private final ApplicationEventPublisher eventPublisher; public void dispatchAlarm(String deviceSerial, int channel, int eventType, MapString, Object extraData) { DeviceEvent event new DeviceEvent(); event.setDeviceSerial(deviceSerial); event.setChannel(channel); event.setEventType(eventType); event.setTimestamp(System.currentTimeMillis()); event.setExtraData(extraData); // 异步发布避免阻塞SDK回调线程 eventPublisher.publishEvent(event); } }需要说明的是这里dispatchAlarm方法是在大华SDK的报警回调里直接调用的所以它在SDK的原生线程上执行。我给它标注了极短的时间复杂度保证只负责创建对象和发布事件。Spring事件监听器异步接收并推送Component public class AlarmPushListener { private final SimpMessagingTemplate messagingTemplate; Async EventListener public void onDeviceEvent(DeviceEvent event) { MapString, Object payload new HashMap(); payload.put(deviceSerial, event.getDeviceSerial()); payload.put(channel, event.getChannel()); payload.put(eventType, event.getEventType()); payload.put(timestamp, event.getTimestamp()); // 推送到所有订阅设备报警主题的终端 messagingTemplate.convertAndSend(/topic/alarm/ event.getDeviceSerial(), payload); } }这里我用的是前面案例二搭好的WebSocket STOMP通道。监控大屏前端只需要在启动时订阅对应设备的报警主题报警事件就能实时上屏。预览流地址的推送逻辑也是类似的思路。用户点击“预览”按钮时请求到达Spring的Controller应用调用大华SDK的实时预览接口拿到RTSP取流URL或者HCVWSS协议地址然后通过WebSocket推送给当前会话。云台控制则反过来前端通过STOMP的send指令把方向和步长发给服务端服务端调用大华SDK的云台控制接口下发给设备。这正好用上了WebSocket双向通信的能力单向SSE在这里是彻底的死路。4.4 大华SDK集成的三个典型坑点这个案例在线上的运维过程中我踩过大华SDK相关的三个坑都很有代表性。第一个是SDK初始化全局唯一。大华SDK的NET_DVR_Init()方法必须在应用启动时调用一次并且全局只能初始化一次。如果多个线程并发调用初始化会导致SDK内部状态混乱轻则设备登录失败重则进程崩溃。我的做法是用一个PostConstruct方法在应用启动阶段显式初始化同时通过一个静态标志位防止重复初始化。第二个是设备登录句柄需要Careful管理。SDK登录后返回一个用户句柄后续所有操作预览、云台控制、报警布防都依赖这个句柄。句柄用完后必须调用登出接口释放否则句柄数量过多会导致SDK资源耗尽。但句柄释放的时机很讲究报警布防的回调还在进行中句柄被释放就会引发回调线程访问野指针导致JVM崩溃。我的经验是给每个句柄加引用计数所有业务操作结束后才真正释放。第三个是JNA内存释放。大华SDK的Java接口通过JNA调用的方式传递结构体时如果涉及指针类型的参数必须在调用完成后手动释放内存。我遇到过一个情况长时间跑下来内存不断增长用MAT分析堆没发现问题后来发现是JNA直接从native堆分配的DirectByteBuffer没有被释放。这个问题的排查非常隐蔽解决方式是在SDK调用工具类里显式调用Pointer.release()或者用try-finally保证释放逻辑一定执行。4.5 推送通道的安全加固从actuator暴露说起的鉴权必要性提到actuator很多人第一时间想到的是生产环境忘了关actuator端点结果内部信息被扫出来。这个思路放在WebSocket推送通道上完全同理。一个只校验了连接来源、没校验用户身份的WebSocket端点就跟对外开放的actuator一样危险——任何人都能订阅/topic/alarm/*主题监控信息直接泄露任何人都能往云台控制主题发指令设备就会被非法操控。我给这个项目加的保护分两层。第一层是握手阶段鉴权所有WebSocket连接必须带有效的JWT这跟case二里的拦截器是一致的。第二层是主题级别授权在ChannelInterceptor里对进入的SUBSCRIBE帧做权限校验只允许用户订阅自己权限范围内的主题。代码要点是拦截MessageType.SUBSCRIBE类型的消息解析目标主题再结合当前Principal判断权限Override public Message? preSend(Message? message, MessageChannel channel) { StompHeaderAccessor accessor StompHeaderAccessor.wrap(message); if (MessageType.SUBSCRIBE.equals(accessor.getCommandType())) { String destination accessor.getDestination(); Principal principal accessor.getUser(); if (!permissionService.canSubscribe(principal, destination)) { throw new AccessDeniedException(无权订阅此主题); } } return message; }5. 常见问题与性能调优实录5.1 实时推送高频问题速查表我在几个项目里积累了一份问题清单基本上新人在实时推送上踩的坑都集中在这张表里现象根因解决思路SSE连接几十秒后自动断开默认超时时间或代理层空闲超时设置SseEmitter无超时加30秒心跳WebSocket能握手但订阅后收不到消息端点前缀或订阅主题路径不一致检查/app、/topic、/user三个前缀的分工广播消息部分用户收不到集群部署但没配StompBrokerRelay引入RabbitMQ或更换外部broker用户多端登录时定向消息只发到一端SendToUser默认发到单一会话用SimpMessagingTemplate指定sessionId消息推送偶尔乱序业务在事务提交前推送使用TransactionalEventListener(AFTER_COMMIT)Tomcat端口溢出长连接没释放确保onCompletion/onTimeout/onError三清理齐全服务端重启后客户端不自动恢复需要客户端重连机制SockJS自带重连SSE用EventSource的onerror处理这张表在技术分享会上反复用过大家可以对照排查。5.2 用Actuator和Micrometer给推送连接装上仪表盘实时推送功能上线后运维最关心的问题就是现在有多少个活跃连接消息推送速率是多少失败率有没有升高这部分我用Actuator加Micrometer做了运行指标监控。Metal监控的核心是把推送通道的实时状态暴露成Metrics。我做的是在SseConnectionManager里维护一个activeConnectionCount的AtomicLong字段用Micrometer的Gauge注册到MeterRegistryComponent public class PushMetrics { public PushMetrics(MeterRegistry registry, SseConnectionManager connectionManager) { Gauge.builder(push.sse.active.connections, connectionManager, SseConnectionManager::getActiveConnectionCount) .description(活跃的SSE连接数) .register(registry); } }然后在application.yaml里开启actuator的metrics端点配合Prometheus采集management: endpoints: web: exposure: include: health,info,metrics,prometheus metrics: export: prometheus: enabled: true这样Grafana上就能实时看到活跃连接数、WebSocket会话数、每分钟推送消息数。我还加了基于连接数的告警规则单节点活跃连接超过5000就预警因为这说明容量接近瓶颈该考虑扩容了。5.3 线程池隔离与前端重连策略推送功能对线程资源的消耗和普通HTTP请求完全不同最怕的是被其他耗时接口拖垮。我在项目中专门为推送逻辑配置了独立的异步线程池核心线程数、最大线程数、队列容量都跟业务线程池分开。这样即使某个业务的慢SQL把默认线程池撑爆了推送线程池依然能正常工作保证用户能实时收到报警。前端重连策略也是实时推送体验的重要一环。SSE里EventSource自带自动重连但重连间隔是浏览器默认的大约3秒服务器繁忙时会造成重连风暴。我建议服务端在发送错误时主动返回一个特殊事件RECONNECT_DELAY客户端收到后手动设置重连延迟做指数退避。WebSocket的客户端就要自己做重连逻辑了我的做法是在onclose回调里用setTimeout重连间隔从1秒开始每次翻倍最多30秒封顶避免服务端故障恢复瞬间所有客户端同时砸过来。另外一个小技巧服务端推送消息前尽量先做一次session.isOpen()检查避免往已经关闭的会话发数据。WebSocket会因为心跳失败、网络闪断等各种原因在服务端还没感知时就失效了主动检查能减少大量无效尝试。实时推送做了这么多年我的体会是它从来都不是一个单一注解或单一框架能搞定的事情。SSE也好、WebSocket也好甚至后面再接上RabbitMQ、Redis、大华SDK本质都是在解决同一个问题让数据在正确的时间以正确的姿态到达正确的终端。技术选型只是第一步连接管理、鉴权隔离、断线重连、集群扩展、监控告警每一项都需要在实际项目里真刀真枪地磨一遍。希望这三个案例能给正在做或准备做实时推送的朋友提供一个完整的参考少走几步弯路。
