Spring 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);
        };
    }
}

@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);
    }
}

@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));
    }
}