公司网页网站如何做,外贸自己做网站,青州市城乡建设局网站,wordpress注册取消邮箱验证SSE详细介绍传送门#xff1a;SSE实时消息推送
简单描述一下SSE推送在实际项目中应用的常见场景 1#xff0c;项目页面中有消息通知板块#xff0c;当信息有变化时#xff0c;只有手动刷新页面#xff0c;才会看到最新的数据#xff0c;这里可以采用SSE技术实时推送最新…SSE详细介绍传送门SSE实时消息推送
简单描述一下SSE推送在实际项目中应用的常见场景 1项目页面中有消息通知板块当信息有变化时只有手动刷新页面才会看到最新的数据这里可以采用SSE技术实时推送最新消息 . 2大屏数据这种场景是可以用SSE进行推送但是需要注意的是SSE是单向的服务端向前端推数据一般要求的是大屏基本没有查询框条件这种比较合适。 注意点如果对于实时数据要求很高并且连接要求做到安全稳定这里推荐用WebSocket一般来说对于数据量小并发连接不是很高要求的情况下SSE足够用而且SSE的配置对于前后端都比较简单但是WebSocket的配置对于后端来说需要花费比较多的时间去完善而且WebSocket是比较消耗服务器资源和网络带宽资源的另外一个如果项目中运维配置了代理服务器的话可能代理服务器也要配置一些支持WebSocket的属性总体来说WebSocket配置的位置比较多容易出现各种坑bug这里注意一下即可。
话不多说总结一下Springboot整合SSE需要的步骤如下: 1编写SSE的服务类主要包括建立连接、关闭连接、异常连接、心跳检测、推送消息等 . 2controller层写入SSE连接和关闭接口 . 3在所需要的业务模块中直接调用SSE服务类中推送消息功能即可 SSE步骤简单无需导入maven依赖踩坑bug少主要是SSE内部支持断线重连爽爽爽
1SSE服务类
package com.bosera.salesioc.home.sse;
import com.alibaba.fastjson.JSONObject;
import com.bosera.salesioc.domain.home.vo.MessageVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;Slf4j
Component
public class SseEmitterServer{private static final ConcurrentHashMapString, MapString,SseEmitter sseEmitterPool new ConcurrentHashMap();private static final ConcurrentHashMapString, Timer headerPool new ConcurrentHashMap();public static ConcurrentHashMapString, MapString, SseEmitter getSseEmitterPool(){return sseEmitterPool;}/*** 建立连接*/public SseEmitter connect(String userCode, String userId){log.info(******************开始建立连接*****************);//设置超时时间0表示不过期默认是30秒超过时间未完成会抛出异常SseEmitter sseemitter new SseEmitter(0L);//注册回调sseemitter.onCompletion(completionCallBack(userCode,userId));sseemitter.onError(errorCallBack(userCode,userId));sseemitter.onTimeout(timeoutCallBack(userCode,userId));sseEmitterPool.computeIfAbsent(userCode, k - new ConcurrentHashMap()).put(userId, sseemitter);// 开启心跳活跃startHeartbeat(sseemitter,userId);return sseemitter;}/*** 关闭当前连接*/public void complete(String userCode, String userId){MapString, SseEmitter map sseEmitterPool.get(userCode);if (map ! null)map.get(userId).complete();}/*** 关闭所有连接*/public void completeAll(){if(!sseEmitterPool.isEmpty()){for (Map.EntryString, MapString, SseEmitter entry : sseEmitterPool.entrySet()) {MapString, SseEmitter userIdMap entry.getValue();if(!userIdMap.isEmpty()){for (Map.EntryString, SseEmitter userIdEntry : userIdMap.entrySet()) {userIdEntry.getValue().complete();}}}sseEmitterPool.clear();}}private Runnable completionCallBack(String userCode, String userId) {return () - {removeUser(userCode,userId);log.info({}结束连接:{},userCode,userId);};}private Runnable timeoutCallBack(String userCode, String userId){return ()-{removeUser(userCode,userId);log.error({}连接超時:{},userCode,userId);};}private ConsumerThrowable errorCallBack(String userCode, String userId){return throwable - {log.error({}连接异常:{},userCode,userId);stopHeartbeat(userId);};}/*** 推送消息*/public void sendMessage(String userCode, MessageVO message){MapString, SseEmitter map sseEmitterPool.get(userCode);if (map ! null) {for (Map.EntryString, SseEmitter entry : map.entrySet()) {try {// 发送事件entry.getValue().send(JSONObject.toJSONString(message));}catch (Exception e){log.error({}连接信息:{}, 错误消息:{},userCode,entry.getKey(),e.getMessage());}}}}private void removeUser(String userCode, String userId){try {MapString, SseEmitter map sseEmitterPool.get(userCode);if (map ! null) {map.remove(userId);// 如果该用户的所有会话都已关闭则移除整个映射if (map.isEmpty())sseEmitterPool.remove(userCode);}// 关闭心跳stopHeartbeat(userId);}catch (Exception e){log.error(关闭连接异常{},e.getMessage());}}/*** 开启心跳*/public void startHeartbeat(SseEmitter sseemitter, String userId) {Timer heartbeatTimer new Timer();headerPool.put(userId,heartbeatTimer);heartbeatTimer.schedule(new TimerTask() {Overridepublic void run() {if (Objects.nonNull(headerPool.get(userId))) {// 发送心跳:保持长连接try {sseemitter.send(connect active);} catch (Exception e) {log.error(connect active error);}}}}, 25000, 25000);}/*** 关闭心跳* param userId*/public void stopHeartbeat(String userId) {Timer timer headerPool.get(userId);if (timer! null)timer.cancel();headerPool.remove(userId);}
}
推送的消息可以统一定义一个类来封装信息 2消息推送响应体
/*** Author xiaozq* Date 2024/2/21* Description: 消息推送响应体*/
public class MessageVOT {// 主题不同位置推送的内容不同private String topic;// 推送消息private T data;public void setTopic(String topic) {this.topic topic;}public void setData(T data) {this.data data;}public String getTopic() {return topic;}public T getData() {return data;}
}3controller层编写连接和关闭接口
RestController
RequestMapping(/sse)
Slf4j
public class SSEController{Autowiredprivate SseEmitterServer sseEmitterServer;/*** 用于创建连接*/GetMapping(value /connect/{userCode}/{userId},produces MediaType.TEXT_EVENT_STREAM_VALUE)public SseEmitter connect(PathVariable(userCode) String userCode, PathVariable(userId) String userId){return sseEmitterServer.connect(userCode, userId);}/*** 关闭连接*/GetMapping(value /close/{userCode}/{userId})public void close(PathVariable(userCode) String userCode,PathVariable(userId) String userId ) {sseEmitterServer.complete(userCode, userId);}}4业务中实际应用:推送消息
Autowired
SseInfoService sseInfoService;
private void handlerMessageInform() {ConcurrentHashMapString, MapString, SseEmitter sessionPool SseEmitterServer.getSseEmitterPool();for (Map.EntryString, MapString, SseEmitter entry : sessionPool.entrySet()) {// 封装消息MessageVOListMessageNotificationVO messageVO new MessageVO();messageVO.setTopic(TopicTypeEnum.MESSAGE_INFORM.getTopic());messageVO.setData(messageService.getMessageList(request));// 推送消息sseEmitterServer.sendMessage(entry.getKey(), messageVO);}}在实践过程中存在的问题 1报错504 gateway timeout这里主要是原项目中配置了响应超时时间不支持长连接这里的做法是心跳活跃保证连接不会被掐断可以写一个定时任务每天晚上定时去关闭所有连接第二天用新的连接这样可以尽量保证内存的连接数不会过多占用内存因为夜深人静的时候谁还会打开web项目工作啊哈哈太卷了吧所以把时间定在晚上最好。 . 如果项目是集群模式的话上述代码就得改造了建议是把消息推送这块单独抽出一个微服务模块来这样子保证所有的连接统一走单独的一个服务因为SSE不是双向的既然是单项连接与后端集群下的其中一个服务建立连接产生的IO流这是只属于当前服务的本地IO关闭IO只能连接对应的这台服务去关闭否则关闭失效。总之考虑的点还有很多一般情况下SSE够用啦 总体来说应用是比较简单的涉及到消息实时推送相关的业务可以尝试SSE
