diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterDelayed.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterDelayed.java new file mode 100644 index 000000000..76d277479 --- /dev/null +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterDelayed.java @@ -0,0 +1,94 @@ +package org.dromara.common.sse.core; + +import lombok.Getter; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.lang.ref.WeakReference; +import java.time.Instant; +import java.util.Objects; +import java.util.concurrent.Delayed; +import java.util.concurrent.TimeUnit; + +/** + * SseEmitterDelayed 是一个用于延迟队列 (DelayQueue) 的包装类， + * 用于管理 SSE (Server-Sent Events) 连接的超时处理 + *

+ * 核心思路： + * - 每个用户的 SSE 连接用一个 WeakReference 包装，避免内存泄漏 + * - 实现 Delayed 接口，允许放入 DelayQueue，根据 expireAt 控制超时 + * - 可在 DelayQueue 中统一处理过期的 SSE 连接，释放资源 + * + * @author AprilWind + */ +@Getter +public class SseEmitterDelayed implements Delayed { + + /** + * 用户ID，用于标识该 SSE 连接属于哪个用户 + */ + private final Long userId; + + /** + * 用于标识 SSE 连接的唯一令牌，通常用于前端校验 + */ + private final String token; + + /** + * SSE 连接的弱引用，避免延迟队列持有强引用导致内存泄漏 + */ + private final WeakReference emitterRef; + + /** + * 该连接的过期时间戳（毫秒），到达该时间后可视为超时 + */ + private final long expireAt; + + /** + * 构造函数 + * + * @param userId 用户ID + * @param token 连接的唯一标识 + * @param emitter 对应的 SseEmitter 对象 + * @param delay 延迟时长 + * @param unit 延迟单位 + */ + public SseEmitterDelayed(Long userId, String token, SseEmitter emitter, long delay, TimeUnit unit) { + this.userId = userId; + this.token = token; + this.emitterRef = new WeakReference<>(emitter); + // expireAt = 当前时间 + 延迟时间 + this.expireAt = Instant.now().toEpochMilli() + unit.toMillis(delay); + } + + /** + * 获取剩余延迟时间 + * + * @param unit 时间单位 + * @return 剩余时间 + */ + @Override + public long getDelay(TimeUnit unit) { + return unit.convert(expireAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS); + } + + /** + * 按 expireAt 时间排序，DelayQueue 内部使用 + * + * @param o 另一个 Delayed 对象 + * @return 比较结果 + */ + @Override + public int compareTo(Delayed o) { + return Long.compare(this.expireAt, ((SseEmitterDelayed) o).expireAt); + } + + /** + * 重写 hashCode 方法，基于 userId 和 token + * 保证在集合中正确比较 + */ + @Override + public int hashCode() { + return Objects.hash(userId, token); + } + +} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java index bc19460f8..563a89dd3 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java @@ -2,13 +2,15 @@ package org.dromara.common.sse.core; import cn.hutool.core.map.MapUtil; import lombok.extern.slf4j.Slf4j; +import org.dromara.common.core.utils.SpringUtils; import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.sse.dto.SseMessageDto; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; +import java.lang.ref.WeakReference; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.*; import java.util.function.Consumer; /** @@ -24,7 +26,63 @@ public class SseEmitterManager { */ private final static String SSE_TOPIC = "global:sse"; - private final static Map> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>(); + private final static Map>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>(); + + /** + * 延迟检测队列 + */ + private static final DelayQueue DELAY_QUEUE = new DelayQueue<>(); + + /** + * 清理线程池 + */ + private static final ScheduledExecutorService CLEANER = + Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "SSE-Cleaner"); + t.setDaemon(true); + return t; + }); + + static { + // 每分钟处理到期任务 + CLEANER.scheduleWithFixedDelay(SseEmitterManager::processDelayQueue, 1, 1, TimeUnit.MINUTES); + } + + /** + * 处理延迟队列中的到期 SSE 任务 + * 作用： + * 1. 移除已经失效的 SSEEmitter + * 2. 对存活的 SSEEmitter 延迟续期，保证心跳检测 + */ + private static void processDelayQueue() { + try { + // 获取当前时间戳，用于判断任务是否到期 + long now = System.currentTimeMillis(); + SseEmitterDelayed task; + + // 循环处理队列中已经到期的任务 + // peek() 获取队首任务，但不移除 + while ((task = DELAY_QUEUE.peek()) != null && task.getExpireAt() <= now) { + // 队首任务已到期，真正取出 + DELAY_QUEUE.poll(); // 从队列中移除任务 + + WeakReference ref = task.getEmitterRef(); + SseEmitter emitter = ref.get(); + + if (emitter == null || isEmitterDead(emitter)) { + // Emitter 已被 GC 或已关闭，断开连接并从管理器移除 + SpringUtils.getBean(SseEmitterManager.class).disconnect(task.getUserId(), task.getToken()); + } else { + // Emitter 仍然存活，延迟续期 + // 5 分钟后再检查该连接，避免频繁触发 + DELAY_QUEUE.offer(new SseEmitterDelayed(task.getUserId(), task.getToken(), emitter, 5, TimeUnit.MINUTES)); + } + } + } catch (Exception e) { + log.error("SSE延迟队列处理异常", e); + } + } + /** * 建立与指定用户的 SSE 连接 @@ -34,41 +92,34 @@ public class SseEmitterManager { * @return 返回一个 SseEmitter 实例，客户端可以通过该实例接收 SSE 事件 */ public SseEmitter connect(Long userId, String token) { + if (userId == null || token == null) { + throw new IllegalArgumentException("userId and token cannot be null"); + } // 从 USER_TOKEN_EMITTERS 中获取或创建当前用户的 SseEmitter 映射表（ConcurrentHashMap） // 每个用户可以有多个 SSE 连接，通过 token 进行区分 - Map emitters = USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>()); + Map> emitters = + USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>()); // 创建一个新的 SseEmitter 实例，超时时间设置为一天 避免连接之后直接关闭浏览器导致连接停滞 SseEmitter emitter = new SseEmitter(86400000L); - emitters.put(token, emitter); + emitters.put(token, new WeakReference<>(emitter)); - // 当 emitter 完成、超时或发生错误时，从映射表中移除对应的 token - emitter.onCompletion(() -> { - SseEmitter remove = emitters.remove(token); - if (remove != null) { - remove.complete(); - } - }); - emitter.onTimeout(() -> { - SseEmitter remove = emitters.remove(token); - if (remove != null) { - remove.complete(); - } - }); - emitter.onError((e) -> { - SseEmitter remove = emitters.remove(token); - if (remove != null) { - remove.complete(); - } - }); + // 三种事件统一处理 + Runnable removeTask = () -> disconnect(userId, token); + emitter.onCompletion(removeTask); + emitter.onTimeout(removeTask); + emitter.onError(e -> removeTask.run()); + + // 延迟清理 + DELAY_QUEUE.offer(new SseEmitterDelayed(userId, token, emitter, 5, TimeUnit.MINUTES)); try { // 向客户端发送一条连接成功的事件 emitter.send(SseEmitter.event().comment("connected")); } catch (IOException e) { - // 如果发送消息失败，则从映射表中移除 emitter - emitters.remove(token); + log.warn("SSE连接发送初始事件失败 userId={}, token={}", userId, token, e); + disconnect(userId, token); } return emitter; } @@ -83,17 +134,40 @@ public class SseEmitterManager { if (userId == null || token == null) { return; } - Map emitters = USER_TOKEN_EMITTERS.get(userId); - if (MapUtil.isNotEmpty(emitters)) { - try { - SseEmitter sseEmitter = emitters.get(token); - sseEmitter.send(SseEmitter.event().comment("disconnected")); - sseEmitter.complete(); - } catch (Exception ignore) { - } - emitters.remove(token); - } else { + Map> emitters = USER_TOKEN_EMITTERS.get(userId); + if (MapUtil.isEmpty(emitters)) { USER_TOKEN_EMITTERS.remove(userId); + return; + } + + WeakReference ref = emitters.remove(token); + if (ref != null) { + SseEmitter emitter = ref.get(); + if (emitter != null) { + try { + emitter.send(SseEmitter.event().comment("disconnected")); + emitter.complete(); + } catch (Exception ignore) { + } + } + } + + if (emitters.isEmpty()) { + USER_TOKEN_EMITTERS.remove(userId); + } + + log.debug("SSE连接移除并断开 userId={}, token={}", userId, token); + } + + /** + * 心跳检测 Emitter 是否已关闭 + */ + private static boolean isEmitterDead(SseEmitter emitter) { + try { + emitter.send(SseEmitter.event().comment("ping")); + return false; + } catch (Exception e) { + return true; } } @@ -113,23 +187,24 @@ public class SseEmitterManager { * @param message 要发送的消息内容 */ public void sendMessage(Long userId, String message) { - Map emitters = USER_TOKEN_EMITTERS.get(userId); - if (MapUtil.isNotEmpty(emitters)) { - for (Map.Entry entry : emitters.entrySet()) { - try { - entry.getValue().send(SseEmitter.event() - .name("message") - .data(message)); - } catch (Exception e) { - SseEmitter remove = emitters.remove(entry.getKey()); - if (remove != null) { - remove.complete(); - } - } - } - } else { + Map> emitters = USER_TOKEN_EMITTERS.get(userId); + if (MapUtil.isEmpty(emitters)) { USER_TOKEN_EMITTERS.remove(userId); + return; } + emitters.forEach((token, ref) -> { + SseEmitter emitter = ref.get(); + if (emitter != null) { + try { + emitter.send(SseEmitter.event().name("message").data(message)); + } catch (Exception e) { + log.warn("SSE消息发送失败 userId={}, token={}", userId, token, e); + disconnect(userId, token); + } + } else { + disconnect(userId, token); + } + }); } /** @@ -138,9 +213,7 @@ public class SseEmitterManager { * @param message 要发送的消息内容 */ public void sendMessage(String message) { - for (Long userId : USER_TOKEN_EMITTERS.keySet()) { - sendMessage(userId, message); - } + USER_TOKEN_EMITTERS.keySet().forEach(userId -> sendMessage(userId, message)); } /** @@ -149,10 +222,7 @@ public class SseEmitterManager { * @param sseMessageDto 要发布的SSE消息对象 */ public void publishMessage(SseMessageDto sseMessageDto) { - SseMessageDto broadcastMessage = new SseMessageDto(); - broadcastMessage.setMessage(sseMessageDto.getMessage()); - broadcastMessage.setUserIds(sseMessageDto.getUserIds()); - RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> { + RedisUtils.publish(SSE_TOPIC, sseMessageDto, consumer -> { log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}", SSE_TOPIC, sseMessageDto.getUserIds(), sseMessageDto.getMessage()); }); @@ -170,4 +240,5 @@ public class SseEmitterManager { log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message); }); } + }