diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java index 0cf8054ed..3b5e8692e 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java +++ b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java @@ -7,6 +7,7 @@ import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; +import org.springframework.scheduling.annotation.EnableScheduling; /** * SSE 自动装配 @@ -16,6 +17,7 @@ import org.springframework.context.annotation.Bean; @AutoConfiguration @ConditionalOnProperty(value = "sse.enabled", havingValue = "true") @EnableConfigurationProperties(SseProperties.class) +@EnableScheduling public class SseAutoConfiguration { @Bean 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..b6a4a6dfc 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 @@ -4,6 +4,7 @@ import cn.hutool.core.map.MapUtil; import lombok.extern.slf4j.Slf4j; import org.dromara.common.redis.utils.RedisUtils; import org.dromara.common.sse.dto.SseMessageDto; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; @@ -38,6 +39,12 @@ public class SseEmitterManager { // 每个用户可以有多个 SSE 连接,通过 token 进行区分 Map emitters = USER_TOKEN_EMITTERS.computeIfAbsent(userId, k -> new ConcurrentHashMap<>()); + // 关闭已存在的SseEmitter,防止超过最大连接数 + SseEmitter oldEmitter = emitters.remove(token); + if (oldEmitter != null) { + oldEmitter.complete(); + } + // 创建一个新的 SseEmitter 实例,超时时间设置为一天 避免连接之后直接关闭浏览器导致连接停滞 SseEmitter emitter = new SseEmitter(86400000L); @@ -97,6 +104,30 @@ public class SseEmitterManager { } } + /** + * SSE心跳监测,关闭无效连接 + */ + @Scheduled(fixedRate = 60000L) // 每分钟执行一次 + public void sseMonitor() { + for (Map.Entry> userEmitters : USER_TOKEN_EMITTERS.entrySet()) { + Map emitters = userEmitters.getValue(); + if (MapUtil.isNotEmpty(emitters)) { + for (Map.Entry entry : emitters.entrySet()) { + try { + // 向客户端发送心跳 + entry.getValue().send(SseEmitter.event().comment("heartbeat")); + } catch (Exception e) { + log.warn("心跳发送失败, 关闭连接: userId={}, token={}", userEmitters.getKey(), entry.getKey()); + SseEmitter remove = emitters.remove(entry.getKey()); + if (remove != null) { + remove.complete(); + } + } + } + } + } + } + /** * 订阅SSE消息主题,并提供一个消费者函数来处理接收到的消息 *