From eab422854809b2c55045fbdd916295fab89b5c3e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=96=AF=E7=8B=82=E7=9A=84=E7=8B=AE=E5=AD=90Li?=
 <15040126243@163.com>
Date: Thu, 25 Jun 2026 17:23:44 +0800
Subject: [PATCH] =?UTF-8?q?fix=20=E4=BF=AE=E5=A4=8D=20push=20=E6=A8=A1?=
 =?UTF-8?q?=E5=9D=97=20sse/ws=20=E6=B5=8F=E8=A7=88=E5=99=A8=E5=A4=9A?=
 =?UTF-8?q?=E6=A0=87=E7=AD=BE=E9=A1=B5=E4=BA=92=E6=8C=A4=E6=8E=89=E7=BA=BF?=
 =?UTF-8?q?=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../common/push/constant/MessageConstants.java   |  5 +++++
 .../push/core/SseEmitterSessionManager.java      | 16 ++++++++++++++++
 .../push/core/WebSocketSessionManager.java       | 14 ++++++++++++++
 3 files changed, 35 insertions(+)

diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
index cefec0238..f35a4a273 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
@@ -31,4 +31,9 @@ public interface MessageConstants {
      * 心跳响应标识
      */
     String PONG = "pong";
+
+    /**
+     * 同一 token 的新连接替换旧连接时发送给旧连接的控制消息。
+     */
+    String KICKED = "kicked";
 }
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java
index 3b0a0ec43..df6365e5c 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java
@@ -65,6 +65,7 @@ public class SseEmitterSessionManager implements PushSessionManager {
         // 关闭已存在的SseEmitter，防止超过最大连接数
         SseEmitter oldEmitter = emitters.remove(token);
         if (oldEmitter != null) {
+            sendKickedMessage(oldEmitter);
             oldEmitter.complete();
         }
 
@@ -103,6 +104,21 @@ public class SseEmitterSessionManager implements PushSessionManager {
         return emitter;
     }
 
+    /**
+     * 通知旧连接已被同 token 新连接替换。
+     *
+     * @param emitter 旧 SSE 连接
+     */
+    private void sendKickedMessage(SseEmitter emitter) {
+        try {
+            emitter.send(SseEmitter.event()
+                .name("message")
+                .data(MessageConstants.KICKED));
+        } catch (Exception ignore) {
+            // 旧连接可能已断开，忽略通知失败
+        }
+    }
+
     /**
      * 断开指定用户的 SSE 连接
      *
diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java
index 2c8f2251b..1ba699d7a 100644
--- a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java
+++ b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java
@@ -5,6 +5,7 @@ import cn.hutool.core.map.MapUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.utils.ThreadUtils;
 import org.dromara.common.json.utils.JsonUtils;
+import org.dromara.common.push.constant.MessageConstants;
 import org.dromara.common.push.dto.PushDTO;
 import org.dromara.common.push.properties.MessageProperties;
 import org.dromara.common.redis.utils.RedisUtils;
@@ -61,6 +62,7 @@ public class WebSocketSessionManager implements PushSessionManager {
         Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.computeIfAbsent(userId, key -> new ConcurrentHashMap<>());
         // 移除并关闭旧的同token会话，避免重复连接
         WebSocketSession oldSession = sessions.remove(token);
+        sendKickedMessage(oldSession);
         closeSession(oldSession, CloseStatus.NORMAL);
         // 存储新会话
         sessions.put(token, session);
@@ -129,6 +131,18 @@ public class WebSocketSessionManager implements PushSessionManager {
         toRemoveUsers.forEach(USER_TOKEN_SESSIONS::remove);
     }
 
+    /**
+     * 通知旧连接已被同 token 新连接替换。
+     *
+     * @param session 旧 WebSocket 会话
+     */
+    private void sendKickedMessage(WebSocketSession session) {
+        if (session == null || !session.isOpen()) {
+            return;
+        }
+        sendMessage(session, MessageConstants.KICKED);
+    }
+
     /**
      * 订阅消息通道
      * 注册消息消费者，监听Redis消息推送