Springboot整合SSE实现实时消息推送

SSE详细介绍传送门：SSE实时消息推送

简单描述一下SSE推送在实际项目中应用的常见场景

1，项目页面中有消息通知板块，当信息有变化时，只有手动刷新页面，才会看到最新的数据，这里可以采用SSE技术实时推送最新消息
.
2，大屏数据，这种场景是可以用SSE进行推送，但是需要注意的是SSE是单向的服务端向前端推数据，一般要求的是大屏基本没有查询框条件这种，比较合适。

注意点：如果对于实时数据要求很高并且连接要求做到安全稳定，这里推荐用WebSocket，一般来说对于数据量小，并发连接不是很高要求的情况下，SSE足够，用而且SSE的配置对于前后端都比较简单，但是WebSocket的配置对于后端来说需要花费比较多的时间去完善，而且WebSocket是比较消耗服务器资源和网络带宽资源的，另外一个，如果项目中运维配置了代理服务器的话，可能代理服务器也要配置一些支持WebSocket的属性，总体来说WebSocket配置的位置比较多，容易出现各种坑bug，这里注意一下即可。

话不多说，总结一下Springboot整合SSE需要的步骤如下:

1，编写SSE的服务类：主要包括建立连接、关闭连接、异常连接、心跳检测、推送消息等
.
2，controller层写入SSE连接和关闭接口
.
3，在所需要的业务模块中直接调用SSE服务类中推送消息功能即可

SSE步骤简单，无需导入maven依赖，踩坑bug少，主要是SSE内部支持断线重连，爽爽爽

1，SSE服务类

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 ConcurrentHashMap<String, Map<String,SseEmitter>> sseEmitterPool = new ConcurrentHashMap<>();private static final ConcurrentHashMap<String, Timer>  headerPool = new ConcurrentHashMap<>();public  static ConcurrentHashMap<String, Map<String, 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){Map<String, SseEmitter> map = sseEmitterPool.get(userCode);if (map != null)map.get(userId).complete();}/*** 关闭所有连接*/public void completeAll(){if(!sseEmitterPool.isEmpty()){for (Map.Entry<String, Map<String, SseEmitter>> entry : sseEmitterPool.entrySet()) {Map<String, SseEmitter> userIdMap = entry.getValue();if(!userIdMap.isEmpty()){for (Map.Entry<String, 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  Consumer<Throwable> errorCallBack(String userCode, String userId){return throwable -> {log.error("{}连接异常:{}",userCode,userId);stopHeartbeat(userId);};}/*** 推送消息*/public  void sendMessage(String userCode, MessageVO message){Map<String, SseEmitter> map = sseEmitterPool.get(userCode);if (map != null) {for (Map.Entry<String, 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 {Map<String, 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 MessageVO<T> {// 主题：不同位置推送的内容不同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;}
}

3，controller层编写连接和关闭接口

@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() {ConcurrentHashMap<String, Map<String, SseEmitter>> sessionPool = SseEmitterServer.getSseEmitterPool();for (Map.Entry<String, Map<String, SseEmitter>> entry : sessionPool.entrySet()) {// 封装消息MessageVO<List<MessageNotificationVO>> 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