Spring SseEmitter推送消息

SseEmitter

SseEmitter 是 Spring Framework 提供的用于支持 Server-Sent Events（SSE）的类，它允许服务器端向客户端推送事件流，实现服务器到客户端的单向通信。

下面我们来看一下SSE的介绍

SSE（Server-Sent Events，服务器推送事件）是一种用于在客户端和服务器之间建立单向通信的技术。它允许服务器端发送异步消息给客户端，而无需客户端明确地请求数据。SSE 基于 HTTP 协议，使用简单的格式来传输文本数据，通常被用于实时通知、实时更新等场景。

我们知道推送消息还有另一种方式是webscoket，那么SSE和webscoket有什么区别？

1. 协议：SSE 基于 HTTP 协议，而 WebSocket 则是一种独立的协议。SSE 使用简单的文本格式传输数据，而 WebSocket 使用二进制帧进行通信。

2. 双向通信：WebSocket 支持双向通信，客户端和服务器可以同时发送和接收数据。而 SSE 是单向通信，只允许服务器向客户端推送数据，客户端无法发送数据给服务器。

3. 连接状态：WebSocket 使用长连接，保持持久连接，可以实现实时双向通信。而 SSE 是基于短连接，每次请求结束后需要重新建立连接。

4. 兼容性：WebSocket 在大多数现代浏览器中都有良好的支持，但在一些旧版本的浏览器中可能会存在兼容性问题。SSE 在较新的浏览器中也有良好的支持，但在旧版本浏览器中可能不被支持。

5. 使用场景：WebSocket 适用于需要实时双向通信的场景，如聊天应用、实时游戏等。SSE 更适合服务器向客户端单向推送实时数据的场景，如股票报价、实时通知等。

综上所述，SSE 和 WebSocket 都是实现实时通信的技术，但在协议、双向通信、连接状态、兼容性和使用场景等方面存在一些区别。选择使用哪种技术取决于具体的需求和应用场景。

SseEmitter 常用方法

下面是 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推送消息工具类

下面，我就用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

SseEmitter搭配监听器

我们还可以搭配监听器来使用，定义某个事件的监听器，当事件发生，就使用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的事件监听，这个之后讲解