From 57f4eaa952c54ff469def8b0c37fe836a25f47f5 Mon Sep 17 00:00:00 2001 From: AprilWind <2100166581@qq.com> Date: Fri, 17 Oct 2025 21:16:54 +0800 Subject: [PATCH] =?UTF-8?q?update=20=E5=A2=9E=E5=BC=BA=20SSE=20=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E7=AE=A1=E7=90=86=EF=BC=8C=E6=94=AF=E6=8C=81=E5=BB=B6?= =?UTF-8?q?=E8=BF=9F=E7=BB=AD=E6=9C=9F=E5=92=8C=E5=BF=83=E8=B7=B3=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/sse/config/SseProperties.java | 21 +++++++ .../common/sse/core/SseEmitterDelayed.java | 6 +- .../common/sse/core/SseEmitterManager.java | 63 ++++++++++++++----- 3 files changed, 73 insertions(+), 17 deletions(-) diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java index ce4e1732d..41810e8ce 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java +++ b/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; + } 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 index 76d277479..cb9ad4d62 100644 --- 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 @@ -41,7 +41,7 @@ public class SseEmitterDelayed implements Delayed { /** * 该连接的过期时间戳（毫秒），到达该时间后可视为超时 */ - private final long expireAt; + private volatile long expireAt; /** * 构造函数 @@ -60,6 +60,10 @@ public class SseEmitterDelayed implements Delayed { this.expireAt = Instant.now().toEpochMilli() + unit.toMillis(delay); } + public void renew(long delay, TimeUnit unit) { + this.expireAt = System.currentTimeMillis() + unit.toMillis(delay); + } + /** * 获取剩余延迟时间 * 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 563a89dd3..72321f720 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 @@ -1,16 +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.*; +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; /** @@ -33,28 +39,51 @@ public class SseEmitterManager { */ private static final DelayQueue DELAY_QUEUE = new DelayQueue<>(); + /** + * 心跳事件 + */ + private static final SseEmitter.SseEventBuilder PING_EVENT = SseEmitter.event().comment("ping"); + + private final SseProperties PROPERTIES = SpringUtils.getBean(SseProperties.class); + /** * 清理线程池 */ - private static final ScheduledExecutorService CLEANER = + private 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); + @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 延迟续期，保证心跳检测 + * 1. 移除已经失效的 SSEEmitter + * 2. 对存活的 SSEEmitter 延迟续期，保证心跳检测 */ - private static void processDelayQueue() { + private void processDelayQueue() { try { // 获取当前时间戳，用于判断任务是否到期 long now = System.currentTimeMillis(); @@ -71,11 +100,11 @@ public class SseEmitterManager { if (emitter == null || isEmitterDead(emitter)) { // Emitter 已被 GC 或已关闭，断开连接并从管理器移除 - SpringUtils.getBean(SseEmitterManager.class).disconnect(task.getUserId(), task.getToken()); + this.disconnect(task.getUserId(), task.getToken()); } else { - // Emitter 仍然存活，延迟续期 - // 5 分钟后再检查该连接，避免频繁触发 - DELAY_QUEUE.offer(new SseEmitterDelayed(task.getUserId(), task.getToken(), emitter, 5, TimeUnit.MINUTES)); + // 直接更新到期时间，放回队列 + task.renew(PROPERTIES.getHeartbeatSeconds(), PROPERTIES.getUnit()); + DELAY_QUEUE.offer(task); } } } catch (Exception e) { @@ -83,7 +112,6 @@ public class SseEmitterManager { } } - /** * 建立与指定用户的 SSE 连接 * @@ -100,6 +128,9 @@ public class SseEmitterManager { Map> emitters = USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>()); + // 如果已有旧连接，则先断开 + emitters.remove(token); + // 创建一个新的 SseEmitter 实例，超时时间设置为一天 避免连接之后直接关闭浏览器导致连接停滞 SseEmitter emitter = new SseEmitter(86400000L); @@ -112,7 +143,8 @@ public class SseEmitterManager { emitter.onError(e -> removeTask.run()); // 延迟清理 - DELAY_QUEUE.offer(new SseEmitterDelayed(userId, token, emitter, 5, TimeUnit.MINUTES)); + DELAY_QUEUE.offer(new SseEmitterDelayed(userId, token, emitter, + PROPERTIES.getHeartbeatSeconds(), PROPERTIES.getUnit())); try { // 向客户端发送一条连接成功的事件 @@ -155,7 +187,6 @@ public class SseEmitterManager { if (emitters.isEmpty()) { USER_TOKEN_EMITTERS.remove(userId); } - log.debug("SSE连接移除并断开 userId={}, token={}", userId, token); } @@ -164,7 +195,7 @@ public class SseEmitterManager { */ private static boolean isEmitterDead(SseEmitter emitter) { try { - emitter.send(SseEmitter.event().comment("ping")); + emitter.send(PING_EVENT); return false; } catch (Exception e) { return true;