Spring Boot SSE 实时推送完全指南 — SseMessageUtils 使用手册基于 RuoYi-AI 项目ruoyi-common-sse模块提炼可直接复用到任何 Spring Boot 3.x 项目。一、架构概览┌─────────────────────────────────────────────────────────────────┐ │ SSE 架构全景 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────────────┐ ┌─────────────────────┐ │ │ │ 前端 │──▶│ SseController │──▶│ SseEmitterManager │ │ │ │EventSrc │ │ GET /sse/connect │ │ USER_TOKEN_EMITTERS │ │ │ └──────────┘ └──────────────────┘ │ MapuserId, │ │ │ │ Maptoken, │ │ │ │ SseEmitter │ │ │ │ 心跳60s / 超时1天 │ │ │ └──────┬──────────────┘ │ │ │ │ │ ┌──────────────────┐ │ │ │ │ SseTopicListener │◀── Redis Pub/Sub ───────┘ │ │ │ 跨节点广播 │ Topic: global:sse │ │ └──────────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────┐ │ │ │ SseMessageUtils │ 静态工具类业务层唯一入口 │ │ │ sendContent() │ ──── 流式内容推送发 N 次 │ │ │ sendReasoning() │ ──── 思考过程推送发 N 次 │ │ │ sendDone() │ ──── 正常结束发 1 次 │ │ │ sendError() │ ──── 异常结束发 1 次 │ │ │ completeConnection│ ──── 关闭物理连接发 1 次 │ │ │ sendMessage() │ ──── 独立推送不绑定连接 │ │ │ publishMessage() │ ──── 跨服务推送走 Redis │ │ │ publishAll() │ ──── 全员广播走 Redis │ │ │ sendEvent() │ ──── 自定义事件 │ │ └──────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘核心设计思想一次 SSE 连接 一次流式对话的生命周期。连接建立 → 流式推送内容 → 发送结束信号 → 关闭连接。同时支持不绑定连接的独立消息推送走 Redis Pub/Sub 跨节点广播。二、环境要求依赖项说明JDK17Spring Boot3.xRedis必须用于跨节点 Pub/Sub 广播和会话管理Sa-Token认证框架SseController通过它获取用户身份可替换最小配置application.ymlsse:enabled:true# 总开关false 则所有方法静默不执行path:/sse/connect# 前端 EventSource 连接的 URL 路径前端建立连接// 浏览器自动携带 Sa-Token 的 Cookie后端据此识别用户constesnewEventSource(/sse/connect);es.addEventListener(content,eappendOutput(JSON.parse(e.data)));es.addEventListener(reasoning,eshowThinking(JSON.parse(e.data)));es.addEventListener(done,ees.close());es.addEventListener(error,e{alert(JSON.parse(e.data).error);es.close();});es.addEventListener(message,eshowNotification(JSON.parse(e.data)));三、事件协议所有 SSE 事件都遵循SseEventDto结构{event:事件类型,content:内容content/reasoning/mcp_tool 事件使用,reasoningContent:推理内容reasoning 事件使用,error:错误信息error 事件使用,done:true/falsedone 事件使用}事件名含义触发方法contentAI 流式输出片段sendContent()reasoningAI 深度思考过程sendReasoning()done流式输出正常完成sendDone()error发生异常sendError()message通用消息推送sendMessage()/publishMessage()mcp_toolMCP 工具调用进度sendEvent()自定义四、方法详解方法分类┌─── sendContent() 流式输出发N次 │ SSE 连接生命周期 ────┼─── sendReasoning() 思考过程发N次 流式场景专用 │ ├─── sendDone() 正常结束发1次 │ ├─── sendError() 异常结束发1次 │ └─── completeConnection() 物理关闭连接发1次 ┌─── sendMessage(uid, msg) 向指定用户发消息 独立推送 ├─── sendMessage(msg) 本机全员广播 不绑定连接 │ ├─── publishMessage(dto) 跨服务指定用户推送 │ └─── publishAll(msg) 跨服务全员广播 自定义事件 ──────────┼─── sendEvent(uid, dto) 发送任意结构化事件4.1sendContent(Long userId, String content)— 流式内容推送作用把 AI 生成的一个 token / 片段实时推给前端。调用时机AI 流式输出的onPartialResponse回调中每收到一个 token 调用一次。调用次数N 次取决于 AI 输出长度是否走 Redis否直接发给本机 SseEmitter// AI 对话流式输出 — ChatServiceFacade.javaOverridepublicvoidonPartialResponse(StringpartialResponse){SseMessageUtils.sendContent(userId,partialResponse);// 每个 token 调一次}前端接收es.addEventListener(content,(e){constdataJSON.parse(e.data);outputDiv.innerHTMLdata.content;// 逐字追加});4.2sendReasoning(Long userId, String reasoningContent)— 思考过程推送作用推送 AI 深度思考Chain of Thought过程与最终回答分开显示。调用时机深度思考模式下模型推理阶段每收到一个 token 调用一次。调用次数N 次是否走 Redis否// 深度思考模式OverridepublicvoidonReasoningToken(Stringtoken){SseMessageUtils.sendReasoning(userId,token);// 灰色小字显示}前端接收es.addEventListener(reasoning,(e){constdataJSON.parse(e.data);thinkingDiv.innerHTMLdata.reasoningContent;});4.3sendDone(Long userId)— 通知正常结束作用告诉前端流式输出完毕一切正常。调用时机全部内容推送完成后只能调一次。调用次数1 次是否走 Redis否// AI 对话正常结束 — ChatServiceFacade.javaOverridepublicvoidonCompleteResponse(ChatResponsecompleteResponse){SseMessageUtils.sendDone(userId);// ① 告诉前端完了// ... 保存消息到数据库 ...SseMessageUtils.completeConnection(userId,token);// ② 关闭物理连接}前端接收es.addEventListener(done,(e){es.close();// 关闭 EventSource});4.4sendError(Long userId, String error)— 通知异常结束作用告诉前端出错了传递错误信息。调用时机发生异常时只能调一次不能和sendDone同时用。调用次数1 次是否走 Redis否// AI 对话异常 — ChatServiceFacade.javaOverridepublicvoidonError(Throwableerror){SseMessageUtils.sendError(userId,error.getMessage());}前端接收es.addEventListener(error,(e){constdataJSON.parse(e.data);alert(出错了data.error);es.close();});4.5completeConnection(Long userId, String tokenValue)— 关闭物理连接作用从连接池中移除并关闭该用户的 SSE 连接释放资源。调用时机sendDone或sendError之后作为收尾操作。调用次数1 次注意需要tokenValue参数这个 token 是SseController.connect()时通过StpUtil.getTokenValue()获取的调用方需要自行保存。是否走 Redis否// 标准 try-catch-finally 模式try{// ... 流式输出 ...SseMessageUtils.sendDone(userId);}catch(Exceptione){SseMessageUtils.sendError(userId,e.getMessage());}finally{SseMessageUtils.completeConnection(userId,tokenValue);// 无论如何都关闭}4.6sendMessage(Long userId, String message)— 向指定用户推送消息作用向指定用户发送一条通用消息不绑定流式连接生命周期。调用时机任意业务场景需要给用户推送通知时订单状态、审批结果等。是否走 Redis是通过 Pub/Sub 跨节点投递// 发送个人通知SseMessageUtils.sendMessage(userId,您的订单已发货);前端接收es.addEventListener(message,(e){constdataJSON.parse(e.data);// data {event:message, content:您的订单已发货}});4.7sendMessage(String message)— 本机全员广播作用给当前 JVM 实例上所有已连接用户推送消息。是否走 Redis否仅本机生效。// 单机场景下的全员通知SseMessageUtils.sendMessage(系统将于 23:00 进行维护);4.8publishMessage(SseMessageDto dto)— 跨服务指定用户推送作用通过 Redis 向指定用户列表推送消息支持微服务跨节点投递。调用时机微服务架构下消息生产者和 SSE 连接不在同一个服务时。是否走 Redis是// AuthController.java — 用户登录 5 秒后推送欢迎消息scheduledExecutorService.schedule(()-{SseMessageDtodtonewSseMessageDto();dto.setMessage(欢迎登录 RuoYi-AI 后台管理系统);dto.setUserIds(List.of(userId));SseMessageUtils.publishMessage(dto);},5,TimeUnit.SECONDS);4.9publishAll(String message)— 跨服务全员广播作用通过 Redis 向所有在线用户广播消息。调用时机全局通知系统公告、紧急维护。是否走 Redis是// SysNoticeController.java — 管理员发布公告后推送给所有在线用户StringtypedictService.getDictLabel(sys_notice_type,notice.getNoticeType());SseMessageUtils.publishAll([type] notice.getNoticeTitle());4.10sendEvent(Long userId, SseEventDto eventDto)— 自定义事件作用发送任意结构化事件适合内置事件类型不够用的场景。是否走 Redis否// MyMcpClientListener.java — MCP 工具调用进度通知SseMessageUtils.sendEvent(userId,SseEventDto.builder().event(mcp).content({\toolName\:\playwright\,\status\:\running\}).build());// 也可以用内置工厂方法SseMessageUtils.sendEvent(userId,SseEventDto.mcpTool(playwright,done,截图已生成));五、完整场景示例场景 1AI 对话流式输出最常用publicvoidstreamChat(LonguserId,StringtokenValue,StreamingChatModelmodel,ListChatMessagemessages){try{model.chat(messages,newStreamingChatResponseHandler(){OverridepublicvoidonPartialResponse(StringpartialResponse){SseMessageUtils.sendContent(userId,partialResponse);// N 次}});SseMessageUtils.sendDone(userId);// ✅ 1 次}catch(Exceptione){SseMessageUtils.sendError(userId,e.getMessage());// ❌ 1 次}finally{SseMessageUtils.completeConnection(userId,tokenValue);// 1 次}}场景 2一次性结果推送publicvoidpushResult(LonguserId,StringtokenValue,Stringresult){try{SseMessageUtils.sendContent(userId,result);SseMessageUtils.sendDone(userId);}catch(Exceptione){SseMessageUtils.sendError(userId,e.getMessage());}finally{SseMessageUtils.completeConnection(userId,tokenValue);}}场景 3深度思考 流式回答model.chat(messages,newStreamingChatResponseHandler(){OverridepublicvoidonReasoningToken(Stringtoken){SseMessageUtils.sendReasoning(userId,token);// 灰色思考过程}OverridepublicvoidonPartialResponse(Stringtoken){SseMessageUtils.sendContent(userId,token);// 正式回答}});// 最后 sendDone completeConnection场景 4系统通知不绑定连接// 管理员发布公告publicvoidpublishNotice(Stringtitle){SseMessageUtils.publishAll(【系统公告】title);// 所有人收到}// 个人通知publicvoidnotifyUser(LonguserId,Stringmsg){SseMessageUtils.sendMessage(userId,msg);// 指定用户收到}场景 5工作流节点实时进度// 每个节点执行完推送进度publicvoidonNodeComplete(StringnodeUuid,StringnodeName,Stringresult){SseMessageUtils.sendEvent(userId,SseEventDto.builder().event(node_complete).content({\uuid\:\nodeUuid\, \name\:\nodeName\}).build());}六、场景速查表场景用到的方法调用模式AI 对话流式输出sendContent(N次) →sendDone(1次) →completeConnection(1次)单连接生命周期AI 对话出错sendContent(N次) →sendError(1次) →completeConnection(1次)单连接生命周期一次性结果推送sendContent(1次) →sendDone(1次) →completeConnection(1次)单连接生命周期深度思考sendReasoning(N次) →sendContent(N次) →sendDone(1次)思考回答两阶段系统公告publishAll(1次)独立推送不绑定连接个人通知sendMessage(uid, msg)或publishMessage(dto)独立推送MCP 工具进度sendEvent(uid, customDto)多次自定义事件流工作流节点进度sendEvent(uid, dto)每个节点一次自定义事件流七、常见错误错误后果正确做法sendDone之后又sendContent前端已关闭连接收不到done 之后不能再发任何内容sendDone和sendError都调用前端收到两个结束信号逻辑混乱互斥只能二选一忘记completeConnection连接泄漏连接池越来越大用 try-finally 保证一定调用流式中间调completeConnection前端连接断开后续内容丢失只在最后调用sendMessage后调completeConnection不需要sendMessage走 Pub/Sub不占用长连接八、迁移 Checklist拷贝ruoyi-common-sse模块到目标项目确保依赖Spring Boot 3.x Redis Sa-Token或改造认证逻辑配置 YAMLsse.enabledtruesse.path/sse/connect前端建立EventSource连接监听对应事件改造认证SseController.connect()中StpUtil.getTokenValue()和LoginHelper.getUserId()换成你自己的认证方式业务代码按上述模板调用SseMessageUtils即可九、核心源码片段SseEmitterManager — 连接池管理// 多用户多连接管理privatefinalstaticMapLong,MapString,SseEmitterUSER_TOKEN_EMITTERSnewConcurrentHashMap();publicSseEmitterconnect(LonguserId,Stringtoken){MapString,SseEmitteremittersUSER_TOKEN_EMITTERS.computeIfAbsent(userId,k-newConcurrentHashMap());// 关闭旧连接防止连接堆积SseEmitteroldEmitteremitters.remove(token);if(oldEmitter!null)oldEmitter.complete();// 创建新连接超时 1 天SseEmitteremitternewSseEmitter(86400000L);emitters.put(token,emitter);// 自动清理回调emitter.onCompletion(()-emitters.remove(token));emitter.onTimeout(()-emitters.remove(token));emitter.onError(e-emitters.remove(token));returnemitter;}// 60 秒心跳自动踢掉死连接publicvoidsseMonitor(){USER_TOKEN_EMITTERS.forEach((userId,emitterMap)-{emitterMap.entrySet().removeIf(entry-{try{entry.getValue().send(SseEmitter.event().comment(heartbeat));returnfalse;// 正常保留}catch(Exceptionex){entry.getValue().complete();returntrue;// 发送失败移除}});});}SseMessageUtils — 统一入口publicclassSseMessageUtils{// 流式内容直接发给本机连接publicstaticvoidsendContent(LonguserId,Stringcontent){sendEvent(userId,SseEventDto.content(content));}// 完成信号publicstaticvoidsendDone(LonguserId){sendEvent(userId,SseEventDto.done());}// 错误信号publicstaticvoidsendError(LonguserId,Stringerror){sendEvent(userId,SseEventDto.error(error));}// 独立消息走 Redis Pub/Sub 跨节点广播publicstaticvoidsendMessage(LonguserId,Stringmessage){SseMessageDtodtonewSseMessageDto();dto.setMessage(message);dto.setUserIds(Collections.singletonList(userId));MANAGER.publishMessage(dto);// → Redis Pub/Sub}// 关闭物理连接publicstaticvoidcompleteConnection(LonguserId,StringtokenValue){MANAGER.disconnect(userId,tokenValue);}}十、总结SseMessageUtils把 SSE 的复杂性封装为 10 个静态方法业务代码只需关注发什么和什么时候发不用关心连接管理、心跳检测、跨节点广播等底层细节。记住两条核心规则流式场景sendContent(N次)→sendDone/sendError(1次)→completeConnection(1次)通知场景sendMessage/publishMessage/publishAll不需要done/error/completeConnection基于 RuoYi-AI 项目提炼Spring Boot 3.x Redis Sa-Token 技术栈