Spring SseEmitter推送消息

2023-12-19 21:58:44

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的事件监听,这个之后讲解

文章来源:https://blog.csdn.net/qq_36551991/article/details/135085163
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。