fix(notify): SSE 改为单用户多连接, 修复多标签页互相顶替导致的重连死循环

daxpay-platform/daxpay-platform-service/service-notify/src/main/java/cn/daxpay/open/platform/notify/service/notice/NotifySseService.java

@@ -7,42 +7,53 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 /// SSE 实时推送服务(管理在线用户的 SseEmitter)
 ///
-/// 单实例方案: 以 userId 维护本地 emitter 映射; 公告发布时推送给所有在线用户.
+/// 多连接方案: 以 userId 维护一组 emitter, 允许同一用户多标签页/多设备并存, 互不顶替;
+/// 各 emitter 依靠自身生命周期回调(onCompletion/onTimeout/onError)与心跳发送失败自动清理.
 /// 多实例横向扩展时需引入 Redis Pub/Sub 跨实例广播(预留扩展点).
 @Slf4j
 @Service
 public class NotifySseService {
 
-    /// userId -> SseEmitter
-    private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
+    /// userId -> 该用户的全部在线连接(支持多标签页/多设备)
+    private final Map<Long, Set<SseEmitter>> emitters = new ConcurrentHashMap<>();
 
-    /// 建立连接
+    /// 建立连接(同一用户新连接不顶替旧连接, 各自独立存活)
     public SseEmitter connect(Long userId) {
-        // 顶掉旧连接
-        SseEmitter old = emitters.remove(userId);
-        if (old != null) {
-            try {
-                old.complete();
-            } catch (Exception ignored) {
-            }
-        }
-        // 0L 表示不超时, 依靠心跳维持
+        // 永不超时, 依靠心跳维持
         SseEmitter emitter = new SseEmitter(0L);
-        emitters.put(userId, emitter);
-        emitter.onCompletion(() -> emitters.remove(userId, emitter));
-        emitter.onTimeout(() -> emitters.remove(userId, emitter));
-        emitter.onError(e -> emitters.remove(userId, emitter));
+        emitters.computeIfAbsent(userId, k -> ConcurrentHashMap.newKeySet()).add(emitter);
+        // 连接结束/超时/出错时自动从集合移除, 空集合回收 key
+        emitter.onCompletion(() -> removeEmitter(userId, emitter));
+        emitter.onTimeout(() -> removeEmitter(userId, emitter));
+        emitter.onError(e -> removeEmitter(userId, emitter));
         return emitter;
     }
 
-    /// 主动断开
+    /// 从某用户的连接集合中移除单个 emitter, 集合空则回收 key
+    private void removeEmitter(Long userId, SseEmitter emitter) {
+        Set<SseEmitter> set = emitters.get(userId);
+        if (set == null) {
+            return;
+        }
+        set.remove(emitter);
+        // 只在集合确实为空时移除, 避免与并发新增竞争
+        if (set.isEmpty()) {
+            emitters.remove(userId, set);
+        }
+    }
+
+    /// 主动断开该用户的全部连接(预留, 如强制下线场景)
     public void disconnect(Long userId) {
-        SseEmitter emitter = emitters.remove(userId);
-        if (emitter != null) {
+        Set<SseEmitter> set = emitters.remove(userId);
+        if (set == null) {
+            return;
+        }
+        for (SseEmitter emitter : set) {
             try {
                 emitter.complete();
             } catch (Exception ignored) {
@@ -55,25 +66,25 @@ public class NotifySseService {
         if (emitters.isEmpty()) {
             return;
         }
-        emitters.forEach((userId, emitter) -> {
+        emitters.forEach((userId, set) -> sendAll(set, payload));
+    }
+
+    /// 推送给指定用户(个人消息场景)
+    public void publishToUser(Long userId, Object payload) {
+        Set<SseEmitter> set = emitters.get(userId);
+        if (set != null) {
+            sendAll(set, payload);
+        }
+    }
+
+    /// 向一组连接广播负载, 发送失败的逐个移除
+    private void sendAll(Set<SseEmitter> set, Object payload) {
+        for (SseEmitter emitter : set) {
             try {
                 emitter.send(SseEmitter.event().data(payload));
             } catch (IOException e) {
-                emitters.remove(userId, emitter);
+                set.remove(emitter);
             }
-        });
-    }
-
-    /// 推送给指定用户(个人消息场景, 预留)
-    public void publishToUser(Long userId, Object payload) {
-        SseEmitter emitter = emitters.get(userId);
-        if (emitter == null) {
-            return;
-        }
-        try {
-            emitter.send(SseEmitter.event().data(payload));
-        } catch (IOException e) {
-            emitters.remove(userId, emitter);
         }
     }
 
@@ -83,11 +94,14 @@ public class NotifySseService {
         if (emitters.isEmpty()) {
             return;
         }
-        emitters.forEach((userId, emitter) -> {
-            try {
-                emitter.send(SseEmitter.event().comment("heartbeat"));
-            } catch (IOException e) {
-                emitters.remove(userId, emitter);
+        emitters.forEach((userId, set) -> {
+            for (SseEmitter emitter : set) {
+                try {
+                    // 注释行不触发前端 onmessage, 仅保活
+                    emitter.send(SseEmitter.event().comment("heartbeat"));
+                } catch (IOException e) {
+                    set.remove(emitter);
+                }
             }
         });
     }