update 添加 SseEmitterDelayed 类以支持 SSE 连接的延迟管理和超时处理

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) 连接的超时处理
+ * <p>
+ * 核心思路：
+ * - 每个用户的 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<SseEmitter> 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);
+    }
+
+}

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<Long, Map<String, SseEmitter>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
+    private final static Map<Long, Map<String, WeakReference<SseEmitter>>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
+
+    /**
+     * 延迟检测队列
+     */
+    private static final DelayQueue<SseEmitterDelayed> 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<SseEmitter> 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<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>());
+        Map<String, WeakReference<SseEmitter>> 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(() -> {
+        Runnable removeTask = () -> disconnect(userId, token);
-            SseEmitter remove = emitters.remove(token);
+        emitter.onCompletion(removeTask);
-            if (remove != null) {
+        emitter.onTimeout(removeTask);
-                remove.complete();
+        emitter.onError(e -> removeTask.run());
-            }
+
-        });
+        // 延迟清理
-        emitter.onTimeout(() -> {
+        DELAY_QUEUE.offer(new SseEmitterDelayed(userId, token, emitter, 5, TimeUnit.MINUTES));
-            SseEmitter remove = emitters.remove(token);
-            if (remove != null) {
-                remove.complete();
-            }
-        });
-        emitter.onError((e) -> {
-            SseEmitter remove = emitters.remove(token);
-            if (remove != null) {
-                remove.complete();
-            }
-        });
 
         try {
             // 向客户端发送一条连接成功的事件
             emitter.send(SseEmitter.event().comment("connected"));
         } catch (IOException e) {
-            // 如果发送消息失败，则从映射表中移除 emitter
+            log.warn("SSE连接发送初始事件失败 userId={}, token={}", userId, token, e);
-            emitters.remove(token);
+            disconnect(userId, token);
         }
         return emitter;
     }
@@ -83,18 +134,41 @@ public class SseEmitterManager {
         if (userId == null || token == null) {
             return;
         }
-        Map<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
+        Map<String, WeakReference<SseEmitter>> emitters = USER_TOKEN_EMITTERS.get(userId);
-        if (MapUtil.isNotEmpty(emitters)) {
+        if (MapUtil.isEmpty(emitters)) {
+            USER_TOKEN_EMITTERS.remove(userId);
+            return;
+        }
+
+        WeakReference<SseEmitter> ref = emitters.remove(token);
+        if (ref != null) {
+            SseEmitter emitter = ref.get();
+            if (emitter != null) {
                 try {
-                SseEmitter sseEmitter = emitters.get(token);
+                    emitter.send(SseEmitter.event().comment("disconnected"));
-                sseEmitter.send(SseEmitter.event().comment("disconnected"));
+                    emitter.complete();
-                sseEmitter.complete();
                 } catch (Exception ignore) {
                 }
-            emitters.remove(token);
+            }
-        } else {
+        }
+
+        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<String, SseEmitter> emitters = USER_TOKEN_EMITTERS.get(userId);
+        Map<String, WeakReference<SseEmitter>> emitters = USER_TOKEN_EMITTERS.get(userId);
-        if (MapUtil.isNotEmpty(emitters)) {
+        if (MapUtil.isEmpty(emitters)) {
-            for (Map.Entry<String, SseEmitter> entry : emitters.entrySet()) {
+            USER_TOKEN_EMITTERS.remove(userId);
+            return;
+        }
+        emitters.forEach((token, ref) -> {
+            SseEmitter emitter = ref.get();
+            if (emitter != null) {
                 try {
-                    entry.getValue().send(SseEmitter.event()
+                    emitter.send(SseEmitter.event().name("message").data(message));
-                        .name("message")
-                        .data(message));
                 } catch (Exception e) {
-                    SseEmitter remove = emitters.remove(entry.getKey());
+                    log.warn("SSE消息发送失败 userId={}, token={}", userId, token, e);
-                    if (remove != null) {
+                    disconnect(userId, token);
-                        remove.complete();
-                    }
-                }
                 }
             } else {
-            USER_TOKEN_EMITTERS.remove(userId);
+                disconnect(userId, token);
             }
+        });
     }
 
     /**
@@ -138,9 +213,7 @@ public class SseEmitterManager {
      * @param message 要发送的消息内容
      */
     public void sendMessage(String message) {
-        for (Long userId : USER_TOKEN_EMITTERS.keySet()) {
+        USER_TOKEN_EMITTERS.keySet().forEach(userId -> sendMessage(userId, message));
-            sendMessage(userId, message);
-        }
     }
 
     /**
@@ -149,10 +222,7 @@ public class SseEmitterManager {
      * @param sseMessageDto 要发布的SSE消息对象
      */
     public void publishMessage(SseMessageDto sseMessageDto) {
-        SseMessageDto broadcastMessage = new SseMessageDto();
+        RedisUtils.publish(SSE_TOPIC, sseMessageDto, consumer -> {
-        broadcastMessage.setMessage(sseMessageDto.getMessage());
-        broadcastMessage.setUserIds(sseMessageDto.getUserIds());
-        RedisUtils.publish(SSE_TOPIC, broadcastMessage, 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);
         });
     }
+
 }