java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > Spring SseEmitter推送消息

Spring SseEmitter推送消息及常用方法

作者：way_more

SseEmitter 是 Spring Framework 提供的用于支持 Server-Sent Events（SSE）的类,它允许服务器端向客户端推送事件流,实现服务器到客户端的单向通信,这篇文章主要介绍了Spring SseEmitter推送消息,需要的朋友可以参考下

SseEmitter

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

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

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

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

综上所述，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的事件监听，这个之后讲解

到此这篇关于Spring SseEmitter推送消息的文章就介绍到这了,更多相关Spring SseEmitter推送消息内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文