SseEmitter 是 Spring Framework 提供的用于支持 Server-Sent Events（SSE）的类，它允许服务器端向客户端推送事件流，实现服务器到客户端的单向通信。
下面我们来看一下SSE的介绍
SSE（Server-Sent Events，服务器推送事件）是一种用于在客户端和服务器之间建立单向通信的技术。它允许服务器端发送异步消息给客户端，而无需客户端明确地请求数据。SSE 基于 HTTP 协议，使用简单的格式来传输文本数据，通常被用于实时通知、实时更新等场景。
我们知道推送消息还有另一种方式是webscoket，那么SSE和webscoket有什么区别？
协议：SSE 基于 HTTP 协议，而 WebSocket 则是一种独立的协议。SSE 使用简单的文本格式传输数据，而 WebSocket 使用二进制帧进行通信。
双向通信：WebSocket 支持双向通信，客户端和服务器可以同时发送和接收数据。而 SSE 是单向通信，只允许服务器向客户端推送数据，客户端无法发送数据给服务器。
连接状态：WebSocket 使用长连接，保持持久连接，可以实现实时双向通信。而 SSE 是基于短连接，每次请求结束后需要重新建立连接。
兼容性：WebSocket 在大多数现代浏览器中都有良好的支持，但在一些旧版本的浏览器中可能会存在兼容性问题。SSE 在较新的浏览器中也有良好的支持，但在旧版本浏览器中可能不被支持。
使用场景：WebSocket 适用于需要实时双向通信的场景，如聊天应用、实时游戏等。SSE 更适合服务器向客户端单向推送实时数据的场景，如股票报价、实时通知等。
综上所述，SSE 和 WebSocket 都是实现实时通信的技术，但在协议、双向通信、连接状态、兼容性和使用场景等方面存在一些区别。选择使用哪种技术取决于具体的需求和应用场景。
下面是 SseEmitter 类的一些常用方法
SseEmitter()：构造方法，用于创建一个新的 SseEmitter 对象。
SseEmitter(Long timeout)：构造方法，用于创建一个设置了超时时间新的 SseEmitter 对象。
send(Object data)：直接发送一个带有默认事件名称的 SSE 事件给客户端，数据由参数指定。
send(Object data, MediaType mediaType)：发送一个带有指定媒体类型的 SSE 事件给客户端。
onTimeout(Runnable callback)：设置连接超时时的回调函数。
onCompletion(Runnable callback)：设置连接完成时的回调函数。
onError(Consumer callback)：设置发送错误时的回调函数。
complete()：表示 SSE 事件流结束，通知客户端不再有新的事件发送。
下面，我就用SseEmitter来编写一个推送消息给所有在线用户和某个用户的工具类
@Slf4j
@Component
public class SseEmitterUtil {
//保存客户连接
public static Map<String, SseEmitter> userSseEmitters = new HashMap<>();
/**
* 用户上线，开启一个SSE连接
* @param userId
* @return
*/
public SseEmitter subscribe(String userId){
//创建一个超时时间为30秒的SseEmitter对象
SseEmitter sseEmitter = new SseEmitter(30*1000L);
//设置回调函数
sseEmitter.onCompletion(completionCallBack(userId));
sseEmitter.onError(errorCallBack(userId));
sseEmitter.onTimeout(timeoutCallBack(userId));
//缓存起来
userSseEmitters.put(userId,sseEmitter);
return sseEmitter;
}
/**
* 推送消息给某个客户端
* @param userId 用户ID
* @param content 消息体
*/
public static void push(String userId,String content){
SseEmitter sseEmitter = userSseEmitters.get(userId);
if (sseEmitter != null) {
try {
sseEmitter.send(content);
}catch (Exception e){
log.error("用户-{} 推送消息异常：{}",userId,e);
}
}else {
log.info("用户-{} 连接不存在",userId);
}
}
/**
* 推送消息给所有客户端
* @param content 消息体
*/
public static void pushAllUser(String content){
for (SseEmitter emitter : userSseEmitters.values()) {
try {
emitter.send(SseEmitter.event().name("全局消息").data(content));
} catch (Exception e) {
log.error("推送消息异常：{}",e);
}
}
}
/**
* 删除某个客户连接
* @param userId
*/
public static void removeEmitter(String userId) {
SseEmitter sseEmitter = userSseEmitters.get(userId);
if (sseEmitter != null) {
sseEmitter.complete();
userSseEmitters.remove(userId);
}else {
log.info("用户-{} 连接不存在",userId);
}
}
private Runnable completionCallBack(String userId) {
return () -> {
log.info("用户-{} 连接成功", userId);
};
}
/**
* 出现超时，将当前用户缓存删除
* @param userId
* @return
*/
private Runnable timeoutCallBack(String userId) {
return () -> {
log.info("用户-{} 连接超时", userId);
userSseEmitters.remove(userId);
};
}
/**
* 出现异常，将当前用户缓存删除
* @param userId
* @return
*/
private Consumer<Throwable> errorCallBack(String userId) {
return throwable -> {
log.info("用户-{} 连接异常", userId);
userSseEmitters.remove(userId);
};
}
}
我们在登录的时候就可以调用subscribe来建立连接，然后发在线用户公告就可以使用pushAllUser，某个用户通知就使用push
我们还可以搭配监听器来使用，定义某个事件的监听器，当事件发生，就使用SseService 来推送消息
如下，我们定义一个监听器，用于监听用户登录的事件。在监听器中，使用 SseService 向所有在线用户发送一条上线通知。
@Component
public class UserLoginListener {
private final SseEmitterUtil sseService;
@Autowired
public UserLoginListener(SseEmitterUtil sseService) {
this.sseService = sseService;
}
@EventListener
public void onUserLogin(UserLoginEvent event) {
String message = "用户 " + event.getUserId() + " 上线了";
sseService.pushAllUser(message);
}
}
当用户登录时，发布一个 UserLoginEvent 事件，触发 UserLoginListener 中的 onUserLogin 方法，向所有在线用户发送上线通知。
@Service
public class UserService {
private final ApplicationEventPublisher publisher;
@Autowired
public UserService(ApplicationEventPublisher publisher) {
this.publisher = publisher;
}
public void login(String userId) {
// 登录逻辑
publisher.publishEvent(new UserLoginEvent(userId));
}
}
事件监听除了可以使用spring的监听机制，还可以使用redis的事件监听，这个之后讲解