diff --git a/daxpay-platform/daxpay-platform-service/service-notify/src/main/java/cn/daxpay/open/platform/notify/service/notice/NotifySseService.java b/daxpay-platform/daxpay-platform-service/service-notify/src/main/java/cn/daxpay/open/platform/notify/service/notice/NotifySseService.java index fbf475b30..aeb1c3f92 100644 --- a/daxpay-platform/daxpay-platform-service/service-notify/src/main/java/cn/daxpay/open/platform/notify/service/notice/NotifySseService.java +++ b/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 emitters = new ConcurrentHashMap<>(); + /// userId -> 该用户的全部在线连接(支持多标签页/多设备) + private final Map> 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 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 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 set = emitters.get(userId); + if (set != null) { + sendAll(set, payload); + } + } + + /// 向一组连接广播负载, 发送失败的逐个移除 + private void sendAll(Set 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); + } } }); }