Pre Merge pull request !777 from AprilWind/dev-sse

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java

@@ -3,6 +3,8 @@ package org.dromara.common.sse.config;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * SSE 配置项
  *
@@ -12,10 +14,29 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 @ConfigurationProperties("sse")
 public class SseProperties {
 
+    /**
+     * 是否启用 SSE 功能
+     */
     private Boolean enabled;
 
     /**
      * 路径
      */
     private String path;
+
+    /**
+     * 心跳检测间隔，默认秒
+     */
+    private long heartbeatSeconds = 30;
+
+    /**
+     * 清理线程执行间隔，默认秒
+     */
+    private long checkIntervalSeconds = 10;
+
+    /**
+     * 时间单位（秒、分钟等），默认秒
+     */
+    private TimeUnit unit = TimeUnit.SECONDS;
+
 }

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterDelayed.java

@@ -0,0 +1,104 @@
+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 volatile 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 delay 延迟时长（例如 30 表示 30 秒）
+     * @param unit  时间单位（如秒、分钟等）
+     */
+    public void renew(long delay, TimeUnit unit) {
+        this.expireAt = System.currentTimeMillis() + 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

@@ -1,14 +1,22 @@
 package org.dromara.common.sse.core;
 
 import cn.hutool.core.map.MapUtil;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.utils.SpringUtils;
 import org.dromara.common.redis.utils.RedisUtils;
+import org.dromara.common.sse.config.SseProperties;
 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.DelayQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
 
 /**
@@ -24,7 +32,85 @@ 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 SseEmitter.SseEventBuilder PING_EVENT = SseEmitter.event().comment("ping");
+
+    private final SseProperties PROPERTIES = SpringUtils.getBean(SseProperties.class);
+
+    /**
+     * 清理线程池
+     */
+    private final ScheduledExecutorService CLEANER =
+        Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "SSE-Cleaner");
+            t.setDaemon(true);
+            return t;
+        });
+
+    @PostConstruct
+    public void init() {
+        log.info("SSE 管理器启动 -> 检查间隔: {} 秒, 心跳间隔: {} 秒",
+            PROPERTIES.getCheckIntervalSeconds(), PROPERTIES.getHeartbeatSeconds());
+
+        CLEANER.scheduleWithFixedDelay(
+            this::processDelayQueue,
+            PROPERTIES.getCheckIntervalSeconds(),
+            PROPERTIES.getCheckIntervalSeconds(),
+            PROPERTIES.getUnit()
+        );
+    }
+
+    @PreDestroy
+    public void destroy() {
+        CLEANER.shutdownNow();
+        USER_TOKEN_EMITTERS.clear();
+        DELAY_QUEUE.clear();
+        log.info("SSE 管理器已关闭");
+    }
+
+    /**
+     * 处理延迟队列中的到期 SSE 任务
+     * 作用：
+     * 1. 移除已经失效的 SSEEmitter
+     * 2. 对存活的 SSEEmitter 延迟续期，保证心跳检测
+     */
+    private 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 或已关闭，断开连接并从管理器移除
+                    this.disconnect(task.getUserId(), task.getToken());
+                } else {
+                    // 直接更新到期时间，放回队列
+                    task.renew(PROPERTIES.getHeartbeatSeconds(), PROPERTIES.getUnit());
+                    DELAY_QUEUE.offer(task);
+                }
+            }
+        } catch (Exception e) {
+            log.error("SSE延迟队列处理异常", e);
+        }
+    }
 
     /**
      * 建立与指定用户的 SSE 连接
@@ -34,41 +120,38 @@ 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<>());
+
+        // 如果已有旧连接，则先断开
+        emitters.remove(token);
 
         // 创建一个新的 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,
-            SseEmitter remove = emitters.remove(token);
+            PROPERTIES.getHeartbeatSeconds(), PROPERTIES.getUnit()));
-            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,17 +166,39 @@ 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)) {
-            try {
-                SseEmitter sseEmitter = emitters.get(token);
-                sseEmitter.send(SseEmitter.event().comment("disconnected"));
-                sseEmitter.complete();
-            } catch (Exception ignore) {
-            }
-            emitters.remove(token);
-        } else {
             USER_TOKEN_EMITTERS.remove(userId);
+            return;
+        }
+
+        WeakReference<SseEmitter> 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(PING_EVENT);
+            return false;
+        } catch (Exception e) {
+            return true;
         }
     }
 
@@ -113,23 +218,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()) {
-                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 {
             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 +244,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 +253,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 +271,5 @@ public class SseEmitterManager {
             log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message);
         });
     }
+
 }