opensource/RuoYi-Vue-Plus
2
0
Fork 1
mirror of https://gitee.com/dromara/RuoYi-Vue-Plus.git synced 2026-09-18 09:35:29 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

update 增强 SSE 连接管理，支持延迟续期和心跳检测配置

Browse Source
This commit is contained in:
AprilWind 2025-10-17 21:16:54 +08:00
parent 0849c2e1c1
commit 57f4eaa952
3 changed files with 73 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@ -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);
}
/**
* 获取剩余延迟时间
*

63
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java
View File

@ -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<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 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<String, WeakReference<SseEmitter>> 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;