From 842e5c29f9312a3ddbe9bfc71d415dce78d99443 Mon Sep 17 00:00:00 2001 From: Azure <983547216@qq.com> Date: Wed, 24 Jun 2026 08:30:22 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=9A=82=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../common/config/SaTokenConfigure.java | 25 ++-- .../common/dto/ApiResponse.java | 2 +- .../webgamebackend/common/event/SseEvent.java | 56 +++---- .../common/event/SseEventPublisher.java | 41 ++---- .../common/event/SseEventType.java | 15 -- .../common/exception/ErrorCode.java | 2 +- .../handler/GlobalExceptionHandler.java | 95 ++---------- .../webgamebackend/config/SseConfig.java | 16 +- .../webgamebackend/sse/RedisSseListener.java | 16 +- .../sse/SseConnectionManager.java | 138 +++++++----------- .../webgamebackend/sse/SseController.java | 127 +++++----------- 11 files changed, 170 insertions(+), 363 deletions(-) diff --git a/src/main/java/com/webgame/webgamebackend/common/config/SaTokenConfigure.java b/src/main/java/com/webgame/webgamebackend/common/config/SaTokenConfigure.java index 5e51ab9..dd3a8f3 100644 --- a/src/main/java/com/webgame/webgamebackend/common/config/SaTokenConfigure.java +++ b/src/main/java/com/webgame/webgamebackend/common/config/SaTokenConfigure.java @@ -12,14 +12,18 @@ import org.springframework.web.servlet.config.annotation.InterceptorRegistry; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; /** - * [Sa-Token 权限认证] 配置类 + * Sa-Token 鉴权配置 */ @Slf4j @Configuration public class SaTokenConfigure implements WebMvcConfigurer { + @Override public void addInterceptors(InterceptorRegistry registry) { registry.addInterceptor(new SaInterceptor(handler -> { + SaRouter.match(SaHttpMethod.OPTIONS).free(r -> { + }).back(); + SaRouter.match("/**") .notMatch("/actuator/**") .notMatch("/swagger-ui/**") @@ -27,32 +31,21 @@ public class SaTokenConfigure implements WebMvcConfigurer { .notMatch("/v3/api-docs/**") .notMatch("/webjars/**") .notMatch("/error") - .notMatch("/sse/**") .check(r -> StpUtil.checkLogin()); })).addPathPatterns("/**") - .excludePathPatterns("/error", "/sse/**"); + .excludePathPatterns("/error"); } - /** - * CORS 跨域处理策略 - */ @Bean public SaCorsHandleFunction corsHandle() { return (req, res, sto) -> { - res - // 允许指定域访问跨域资源 - .setHeader("Access-Control-Allow-Origin", "*")// 允许指定域访问跨域资源 - // 允许所有请求方式 + res.setHeader("Access-Control-Allow-Origin", "*") .setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS, DELETE, PUT, PATCH") - // 有效时间 .setHeader("Access-Control-Max-Age", "3600") - // 允许的header参数 .setHeader("Access-Control-Allow-Headers", "*"); - // 如果是预检请求，则立即返回到前端 - SaRouter.match(SaHttpMethod.OPTIONS) - .free(r -> System.out.println("--------OPTIONS预检请求，不做处理")) - .back(); + SaRouter.match(SaHttpMethod.OPTIONS).free(r -> { + }).back(); }; } } diff --git a/src/main/java/com/webgame/webgamebackend/common/dto/ApiResponse.java b/src/main/java/com/webgame/webgamebackend/common/dto/ApiResponse.java index 5c16fa0..f047bfb 100644 --- a/src/main/java/com/webgame/webgamebackend/common/dto/ApiResponse.java +++ b/src/main/java/com/webgame/webgamebackend/common/dto/ApiResponse.java @@ -11,7 +11,7 @@ public record ApiResponse(int code, String message, T data) { * 成功响应 */ public static ApiResponse ok(T data) { - return new ApiResponse<>(0, "success", data); + return new ApiResponse<>(0, "成功", data); } /** diff --git a/src/main/java/com/webgame/webgamebackend/common/event/SseEvent.java b/src/main/java/com/webgame/webgamebackend/common/event/SseEvent.java index e8892fc..412a7b9 100644 --- a/src/main/java/com/webgame/webgamebackend/common/event/SseEvent.java +++ b/src/main/java/com/webgame/webgamebackend/common/event/SseEvent.java @@ -1,50 +1,42 @@ package com.webgame.webgamebackend.common.event; import jakarta.annotation.Nullable; +import java.util.Objects; +import java.util.UUID; /** - * SSE 事件模型（不可变 record） + * SSE 事件模型 * - * @param id 事件唯一 ID，格式 "evt_{timestamp}_{random}" + * @param id 事件 ID，客户端用于追踪和断线续传 * @param type 事件类型 - * @param targetId 目标用户 accountId，null 表示广播给所有在线用户 - * @param data 事件携带的 JSON 数据 - * @param timestamp 事件时间戳（毫秒） + * @param targetId 目标账号 ID，null 表示广播 + * @param data JSON 数据载荷 + * @param timestamp 事件创建时间戳（毫秒） */ public record SseEvent( - String id, - SseEventType type, - @Nullable String targetId, - String data, - long timestamp + String id, + SseEventType type, + @Nullable String targetId, + String data, + long timestamp ) { - /** - * 创建广播事件（无特定目标用户） - * - * @param type 事件类型 - * @param data JSON 数据字符串 - * @return SseEvent 实例 - */ + public SseEvent { + Objects.requireNonNull(id, "id 不能为空"); + Objects.requireNonNull(type, "type 不能为空"); + Objects.requireNonNull(data, "data 不能为空"); + } + public static SseEvent broadcast(SseEventType type, String data) { - long now = System.currentTimeMillis(); - return new SseEvent("evt_" + now + "_" + randomSuffix(), type, null, data, now); + return create(type, null, data); } - /** - * 创建定向事件（推送给特定用户） - * - * @param type 事件类型 - * @param targetId 目标用户 accountId - * @param data JSON 数据字符串 - * @return SseEvent 实例 - */ public static SseEvent targeted(SseEventType type, String targetId, String data) { - long now = System.currentTimeMillis(); - return new SseEvent("evt_" + now + "_" + randomSuffix(), type, targetId, data, now); + Objects.requireNonNull(targetId, "targetId 不能为空"); + return create(type, targetId, data); } - /** 生成 6 位随机后缀 */ - private static String randomSuffix() { - return String.format("%06x", (long) (Math.random() * 0xFFFFFF)); + private static SseEvent create(SseEventType type, @Nullable String targetId, String data) { + long now = System.currentTimeMillis(); + return new SseEvent("evt_" + now + "_" + UUID.randomUUID(), type, targetId, data, now); } } diff --git a/src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java b/src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java index ca1c27f..a18b0a2 100644 --- a/src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java +++ b/src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java @@ -7,51 +7,36 @@ import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; /** - * SSE 事件发布器 + * 统一 SSE 事件发布器 * - * 统一入口，将业务事件序列化后发送到 Redis Pub/Sub 频道， - * 由所有实例的 RedisSseListener 接收并推送到各自连接的客户端。 + * 业务代码通过此发布器发布领域无关事件，由 Redis Pub/Sub 扇出到所有应用实例。 */ @Slf4j @Component @RequiredArgsConstructor public class SseEventPublisher { - /** Redis Pub/Sub 频道名 */ public static final String CHANNEL = "sse:events"; private final RedisTemplate redisTemplate; - /** - * 发布广播事件（推送给所有在线用户） - * - * @param type 事件类型 - * @param data 事件数据对象（将被序列化为 JSON） - */ public void broadcast(SseEventType type, Object data) { - SseEvent event = SseEvent.broadcast(type, JSON.toJSONString(data)); - publish(event); + publish(SseEvent.broadcast(type, JSON.toJSONString(data))); } - /** - * 发布定向事件（只推送给指定用户） - * - * @param userId 目标用户 accountId - * @param type 事件类型 - * @param data 事件数据对象（将被序列化为 JSON） - */ public void sendTo(String userId, SseEventType type, Object data) { - SseEvent event = SseEvent.targeted(type, userId, JSON.toJSONString(data)); - publish(event); + publish(SseEvent.targeted(type, userId, JSON.toJSONString(data))); } - /** - * 将事件序列化为 JSON 并发布到 Redis Pub/Sub 频道 - */ private void publish(SseEvent event) { - String json = JSON.toJSONString(event); - log.debug("[SSE] 发布事件到 Redis, channel={}, eventType={}, targetId={}", - CHANNEL, event.type(), event.targetId()); - redisTemplate.convertAndSend(CHANNEL, json); + try { + String json = JSON.toJSONString(event); + redisTemplate.convertAndSend(CHANNEL, json); + log.debug("[SSE] 事件已发布, channel={}, type={}, targetId={}", + CHANNEL, event.type(), event.targetId()); + } catch (Exception e) { + log.error("[SSE] 发布事件失败, type={}, targetId={}", + event.type(), event.targetId(), e); + } } } diff --git a/src/main/java/com/webgame/webgamebackend/common/event/SseEventType.java b/src/main/java/com/webgame/webgamebackend/common/event/SseEventType.java index f4a0f9d..8fe8c18 100644 --- a/src/main/java/com/webgame/webgamebackend/common/event/SseEventType.java +++ b/src/main/java/com/webgame/webgamebackend/common/event/SseEventType.java @@ -2,27 +2,12 @@ package com.webgame.webgamebackend.common.event; /** * SSE 事件类型枚举 - *

- * ROOM_CREATED / ROOM_UPDATED / ROOM_REMOVED 为广播事件（推送给所有在线用户） - * KICKED / GAME_STARTED / BALANCE_CHG 为定向事件（推送给特定用户） */ public enum SseEventType { - - /** 新房间创建（广播） */ ROOM_CREATED, - - /** 房间状态更新：玩家加入/离开、准备切换、房主变更、游戏开始等（广播） */ ROOM_UPDATED, - - /** 房间移除：无人散场或游戏结束关闭（广播） */ ROOM_REMOVED, - - /** 被房主踢出房间（定向） */ KICKED, - - /** 你所在的房间游戏已开始（定向） */ GAME_STARTED, - - /** 余额变动通知（定向） */ BALANCE_CHG } diff --git a/src/main/java/com/webgame/webgamebackend/common/exception/ErrorCode.java b/src/main/java/com/webgame/webgamebackend/common/exception/ErrorCode.java index cffbe0e..706c48b 100644 --- a/src/main/java/com/webgame/webgamebackend/common/exception/ErrorCode.java +++ b/src/main/java/com/webgame/webgamebackend/common/exception/ErrorCode.java @@ -9,7 +9,7 @@ import lombok.Getter; public enum ErrorCode { // ========== 通用 ========== - SUCCESS(0, "success"), + SUCCESS(0, "成功"), BAD_REQUEST(400, "参数校验失败"), INTERNAL_ERROR(5000, "服务器内部错误"), diff --git a/src/main/java/com/webgame/webgamebackend/common/handler/GlobalExceptionHandler.java b/src/main/java/com/webgame/webgamebackend/common/handler/GlobalExceptionHandler.java index c9aabe0..70eaedd 100644 --- a/src/main/java/com/webgame/webgamebackend/common/handler/GlobalExceptionHandler.java +++ b/src/main/java/com/webgame/webgamebackend/common/handler/GlobalExceptionHandler.java @@ -13,150 +13,81 @@ import org.springframework.web.HttpRequestMethodNotSupportedException; import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RestControllerAdvice; +import org.springframework.web.context.request.async.AsyncRequestNotUsableException; import org.springframework.web.context.request.async.AsyncRequestTimeoutException; import org.springframework.web.method.annotation.MethodArgumentTypeMismatchException; -import org.springframework.web.context.request.async.AsyncRequestNotUsableException; import org.springframework.web.servlet.resource.NoResourceFoundException; -/** - * 全局异常处理，统一返回 ApiResponse JSON 格式。 - * - * SSE 端点的 401 由拦截器统一处理后返回 JSON，前端 fetch-event-source 通过 - * err.response.status 读取，不再需要服务端以 SSE 事件格式写错误。 - */ @Slf4j @RestControllerAdvice public class GlobalExceptionHandler { - // ==================== SSE 异步超时 ==================== - - /** - * SSE 异步请求超时（SseEmitter timeout 到期的正常行为） - * - * 返回 void 避免 Spring 尝试写入已提交的响应导致二次异常。 - */ @ExceptionHandler(AsyncRequestTimeoutException.class) public void handleAsyncTimeout(AsyncRequestTimeoutException ex) { - // SSE 长连接超时是正常行为，无需记录日志 + // SSE 超时在长连接场景下属于正常情况 } - /** - * SaToken 上下文未初始化（异步分派时线程局部变量丢失） - * - * 返回 void 避免 Spring 尝试以 JSON 写入已提交的 text/event-stream 响应导致二次异常。 - */ @ExceptionHandler(SaTokenContextException.class) public void handleSaTokenContext(SaTokenContextException ex) { - log.warn("SaToken 上下文未初始化（异步分派场景）: {}", ex.getMessage()); + log.warn("Sa-Token 上下文不可用: {}", ex.getMessage()); } - /** - * 异步请求响应已不可用（SSE 连接断开后心跳试图写入） - * - * 返回 void 避免 Spring 尝试以 JSON 写入已损坏的 text/event-stream 响应。 - */ @ExceptionHandler(AsyncRequestNotUsableException.class) public void handleAsyncRequestNotUsable(AsyncRequestNotUsableException ex) { - // SSE 连接已断开后心跳的残留异常，无需处理 + // SSE 响应已被客户端关闭 } - // ==================== 业务异常 ==================== - - /** - * 处理业务异常 - */ @ExceptionHandler(BusinessException.class) public ApiResponse handleBusinessException(BusinessException ex) { log.warn("业务异常: code={}, message={}", ex.getCode(), ex.getMessage()); return ApiResponse.fail(ex.getCode(), ex.getMessage()); } - // ==================== 参数校验异常 ==================== - - /** - * 处理 @Valid 参数校验失败，合并所有字段的错误消息 - */ @ExceptionHandler(MethodArgumentNotValidException.class) public ApiResponse handleValidation(MethodArgumentNotValidException ex) { String message = ex.getBindingResult().getFieldErrors().stream() - .map(e -> e.getField() + " " + e.getDefaultMessage()) + .map(error -> error.getField() + " " + error.getDefaultMessage()) .reduce((a, b) -> a + "; " + b) .orElse(ErrorCode.BAD_REQUEST.getDefaultMessage()); return ApiResponse.fail(ErrorCode.BAD_REQUEST.getCode(), message); } - // ==================== Sa-Token 鉴权异常 ==================== - - /** - * 未登录 / token 无效 - */ @ExceptionHandler(NotLoginException.class) public ApiResponse handleNotLogin(NotLoginException ex) { log.warn("未登录: {}", ex.getMessage()); return ApiResponse.fail(401, "请先登录"); } - /** - * 无权限 - */ - @ExceptionHandler(NotPermissionException.class) - public ApiResponse handleNotPermission(NotPermissionException ex) { - log.warn("无权限: {}", ex.getMessage()); - return ApiResponse.fail(403, "无权限"); + @ExceptionHandler({NotPermissionException.class, NotRoleException.class}) + public ApiResponse handleForbidden(Exception ex) { + log.warn("权限不足: {}", ex.getMessage()); + return ApiResponse.fail(403, "权限不足"); } - /** - * 无角色 - */ - @ExceptionHandler(NotRoleException.class) - public ApiResponse handleNotRole(NotRoleException ex) { - log.warn("无角色: {}", ex.getMessage()); - return ApiResponse.fail(403, "无权限"); - } - - // ==================== Spring MVC 异常 ==================== - - /** - * 请求方法不支持（如 GET 访问 POST 接口） - */ @ExceptionHandler(HttpRequestMethodNotSupportedException.class) public ApiResponse handleMethodNotSupported(HttpRequestMethodNotSupportedException ex) { log.warn("请求方法不支持: {}", ex.getMessage()); return ApiResponse.fail(405, "请求方法不支持"); } - /** - * 请求体不可读（如 JSON 格式错误） - */ @ExceptionHandler(HttpMessageNotReadableException.class) public ApiResponse handleMessageNotReadable(HttpMessageNotReadableException ex) { log.warn("请求体解析失败: {}", ex.getMessage()); - return ApiResponse.fail(ErrorCode.BAD_REQUEST.getCode(), "请求体格式错误"); + return ApiResponse.fail(ErrorCode.BAD_REQUEST.getCode(), "请求体格式无效"); } - /** - * 路径参数类型不匹配 - */ @ExceptionHandler(MethodArgumentTypeMismatchException.class) public ApiResponse handleTypeMismatch(MethodArgumentTypeMismatchException ex) { log.warn("参数类型不匹配: {}", ex.getMessage()); - return ApiResponse.fail(ErrorCode.BAD_REQUEST.getCode(), "参数类型错误"); + return ApiResponse.fail(ErrorCode.BAD_REQUEST.getCode(), "参数类型无效"); } - /** - * 资源不存在（404） - */ @ExceptionHandler(NoResourceFoundException.class) public ApiResponse handleNoResourceFound(NoResourceFoundException ex) { - log.warn("资源不存在: {}", ex.getMessage()); - return ApiResponse.fail(404, "资源不存在"); + log.warn("资源未找到: {}", ex.getMessage()); + return ApiResponse.fail(404, "资源未找到"); } - // ==================== 兜底 ==================== - - /** - * 处理其他未捕获异常 - */ @ExceptionHandler(Exception.class) public ApiResponse handleException(Exception ex) { log.error("服务器内部错误", ex); diff --git a/src/main/java/com/webgame/webgamebackend/config/SseConfig.java b/src/main/java/com/webgame/webgamebackend/config/SseConfig.java index 02fdc3f..2486b22 100644 --- a/src/main/java/com/webgame/webgamebackend/config/SseConfig.java +++ b/src/main/java/com/webgame/webgamebackend/config/SseConfig.java @@ -7,11 +7,11 @@ import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** - * SSE 相关 Bean 配置 - * - * 注册 Redis Pub/Sub 消息监听器，订阅频道 "sse:events"。 + * SSE 基础设施配置 */ @Configuration public class SseConfig { @@ -25,4 +25,14 @@ public class SseConfig { container.addMessageListener(listener, new ChannelTopic(SseEventPublisher.CHANNEL)); return container; } + + @Bean + public TaskScheduler sseTaskScheduler() { + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(2); + scheduler.setThreadNamePrefix("sse-heartbeat-"); + scheduler.setRemoveOnCancelPolicy(true); + scheduler.initialize(); + return scheduler; + } } diff --git a/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java b/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java index 0c607fa..3418321 100644 --- a/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java +++ b/src/main/java/com/webgame/webgamebackend/sse/RedisSseListener.java @@ -11,10 +11,7 @@ import org.springframework.stereotype.Component; import java.nio.charset.StandardCharsets; /** - * Redis Pub/Sub 消息监听器 - * - * 接收来自 Redis 频道 "sse:events" 的消息， - * 反序列化后根据事件类型（广播/定向）转发给本地 SseConnectionManager。 + * SSE 事件的 Redis Pub/Sub 监听器 */ @Slf4j @Component @@ -26,22 +23,21 @@ public class RedisSseListener implements MessageListener { @Override public void onMessage(Message message, byte[] pattern) { String body = new String(message.getBody(), StandardCharsets.UTF_8); - log.debug("[SSE] 收到 Redis 消息, channel={}", new String(message.getChannel(), StandardCharsets.UTF_8)); try { SseEvent event = JSON.parseObject(body, SseEvent.class); if (event == null) { - log.warn("[SSE] 无法解析事件消息: {}", body); + log.warn("[SSE] 收到空的 Redis 事件, body={}", body); return; } - if (event.targetId() != null) { - connectionManager.sendTo(event.targetId(), event); - } else { + if (event.targetId() == null) { connectionManager.broadcast(event); + } else { + connectionManager.sendTo(event.targetId(), event); } } catch (Exception e) { - log.error("[SSE] 处理 Redis 消息失败: {}", body, e); + log.error("[SSE] 处理 Redis 消息失败, body={}", body, e); } } } diff --git a/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java b/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java index 144e22b..fd61cdf 100644 --- a/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java +++ b/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java @@ -7,132 +7,100 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; import java.util.Map; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; /** - * SSE 连接管理器 + * 线程安全的 SSE 连接管理器 * - * 维护 userId → SseEmitter 映射，提供广播和定向推送能力。 - * 使用 ConcurrentHashMap 保证线程安全。 + * 一个用户可以拥有多个连接，使得不同浏览器标签页和设备之间不会互相踢下线。 */ @Slf4j @Component public class SseConnectionManager { - /** userId → SseEmitter 映射表，线程安全 */ - private final ConcurrentHashMap emitters = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> emitters = + new ConcurrentHashMap<>(); + + public String register(String userId, SseEmitter emitter) { + String connectionId = UUID.randomUUID().toString(); + emitters.computeIfAbsent(userId, key -> new ConcurrentHashMap<>()).put(connectionId, emitter); - /** - * 注册新的 SSE 连接 - * - * @param userId 用户 accountId - * @param emitter SseEmitter 实例 - */ - public void register(String userId, SseEmitter emitter) { - // 如果已有旧连接，先关闭再替换 - SseEmitter old = emitters.put(userId, emitter); - if (old != null) { - try { - old.complete(); - } catch (Exception ignored) { - } - } - // 注册清理回调 emitter.onCompletion(() -> { - log.debug("[SSE] 连接正常关闭, userId={}", userId); - emitters.remove(userId, emitter); + log.debug("[SSE] Connection completed, userId={}, connectionId={}", userId, connectionId); + remove(userId, connectionId); }); emitter.onTimeout(() -> { - log.debug("[SSE] 连接超时, userId={}", userId); - emitters.remove(userId, emitter); + log.debug("[SSE] Connection timeout, userId={}, connectionId={}", userId, connectionId); + remove(userId, connectionId); }); emitter.onError(throwable -> { - log.debug("[SSE] 连接异常, userId={}, reason={}", userId, throwable.getMessage()); - emitters.remove(userId, emitter); + log.debug("[SSE] Connection error, userId={}, connectionId={}, reason={}", + userId, connectionId, throwable.getMessage()); + remove(userId, connectionId); }); - log.info("[SSE] 用户连接成功, userId={}, 当前在线={}", userId, emitters.size()); + log.info("[SSE] User connected, userId={}, connectionId={}, onlineConnections={}", + userId, connectionId, getOnlineCount()); + return connectionId; } - /** - * 移除连接 - * - * 先移除映射再尝试关闭，避免回调递归。 - * 响应已损坏时 complete() 可能抛出 RuntimeException，静默忽略。 - * - * @param userId 用户 accountId - */ - public void remove(String userId) { - SseEmitter removed = emitters.remove(userId); + public void remove(String userId, String connectionId) { + ConcurrentHashMap userEmitters = emitters.get(userId); + if (userEmitters == null) { + return; + } + + SseEmitter removed = userEmitters.remove(connectionId); + if (userEmitters.isEmpty()) { + emitters.remove(userId, userEmitters); + } + if (removed != null) { try { removed.complete(); } catch (Throwable ignored) { - // 响应已损坏（AsyncRequestNotUsableException 等），忽略 + // 客户端断开后响应可能已不可用 } - log.info("[SSE] 用户连接已移除, userId={}, 当前在线={}", userId, emitters.size()); + log.info("[SSE] Connection removed, userId={}, connectionId={}, onlineConnections={}", + userId, connectionId, getOnlineCount()); } } - /** - * 广播事件给所有在线用户 - * - * @param event SSE 事件 - */ public void broadcast(SseEvent event) { - if (emitters.isEmpty()) return; - - log.debug("[SSE] 广播事件, type={}, 接收人数={}", event.type(), emitters.size()); - for (Map.Entry entry : emitters.entrySet()) { - sendEvent(entry.getValue(), event, entry.getKey()); - } - } - - /** - * 发送事件给指定用户 - * - * @param userId 目标用户 accountId - * @param event SSE 事件 - */ - public void sendTo(String userId, SseEvent event) { - SseEmitter emitter = emitters.get(userId); - if (emitter == null) { - log.debug("[SSE] 目标用户不在线, userId={}, eventType={}", userId, event.type()); + if (emitters.isEmpty()) { return; } - log.debug("[SSE] 定向推送, userId={}, eventType={}", userId, event.type()); - sendEvent(emitter, event, userId); + log.debug("[SSE] Broadcast event, type={}, onlineConnections={}", event.type(), getOnlineCount()); + emitters.forEach((userId, connections) -> + connections.forEach((connectionId, emitter) -> sendEvent(userId, connectionId, emitter, event))); + } + + public void sendTo(String userId, SseEvent event) { + ConcurrentHashMap connections = emitters.get(userId); + if (connections == null || connections.isEmpty()) { + log.debug("[SSE] Target user is offline, userId={}, eventType={}", userId, event.type()); + return; + } + + connections.forEach((connectionId, emitter) -> sendEvent(userId, connectionId, emitter, event)); } - /** - * 获取当前在线连接数 - * - * @return 在线用户数 - */ public int getOnlineCount() { - return emitters.size(); + return emitters.values().stream().mapToInt(Map::size).sum(); } - /** - * 向单个 SseEmitter 发送事件 - * - * 捕获 IOException（客户端已断开但还未触发回调）， - * 此时安全移除连接。 - */ - private void sendEvent(SseEmitter emitter, SseEvent event, String userId) { + private void sendEvent(String userId, String connectionId, SseEmitter emitter, SseEvent event) { try { emitter.send(SseEmitter.event() .id(event.id()) .name(event.type().name().toLowerCase()) .data(event.data())); - } catch (IOException e) { - log.debug("[SSE] 推送失败（客户端可能已断开）, userId={}, reason={}", userId, e.getMessage()); - emitters.remove(userId); - try { - emitter.completeWithError(e); - } catch (Exception ignored) { - } + } catch (IOException | IllegalStateException e) { + log.debug("[SSE] Push failed, userId={}, connectionId={}, reason={}", + userId, connectionId, e.getMessage()); + remove(userId, connectionId); } } } diff --git a/src/main/java/com/webgame/webgamebackend/sse/SseController.java b/src/main/java/com/webgame/webgamebackend/sse/SseController.java index 305c7c4..c87ad69 100644 --- a/src/main/java/com/webgame/webgamebackend/sse/SseController.java +++ b/src/main/java/com/webgame/webgamebackend/sse/SseController.java @@ -1,33 +1,22 @@ package com.webgame.webgamebackend.sse; import cn.dev33.satoken.stp.StpUtil; -import com.alibaba.fastjson2.JSON; -import com.webgame.webgamebackend.common.event.SseEvent; -import com.webgame.webgamebackend.common.event.SseEventType; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.http.MediaType; +import org.springframework.scheduling.TaskScheduler; import org.springframework.web.bind.annotation.GetMapping; -import org.springframework.web.bind.annotation.PostMapping; -import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; -import java.util.Map; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; +import java.time.Duration; +import java.util.concurrent.ScheduledFuture; /** - * SSE 连接端点 - * - * 提供 GET /sse/subscribe 端点，建立持久化 SSE 连接。 - * 连接成功后立即发送 connected 事件确认握手，之后每 30 秒发送心跳保活。 - * - * 鉴权由 SaTokenConfigure 拦截器统一处理，前端通过 fetch-event-source - * 的 headers 传 saToken，401 时由前端 err.response.status 直接读取。 + * SSE 订阅端点 */ @Slf4j @RestController @@ -35,93 +24,51 @@ import java.util.concurrent.TimeUnit; @RequiredArgsConstructor public class SseController { + private static final long SSE_TIMEOUT_MS = 30 * 60 * 1000L; + private static final Duration HEARTBEAT_INTERVAL = Duration.ofSeconds(30); + private final SseConnectionManager connectionManager; + private final TaskScheduler sseTaskScheduler; - /** - * 建立 SSE 订阅连接 - * - * 鉴权由拦截器处理，此处直接获取登录用户 ID。 - * - * @return SseEmitter 实例（30 分钟超时） - */ @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE) - public SseEmitter subscribe() { + public SseEmitter subscribe( + @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) { String userId = StpUtil.getLoginIdAsString(); - SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30 分钟超时 + SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS); + String connectionId = connectionManager.register(userId, emitter); - connectionManager.register(userId, emitter); + ScheduledFuture heartbeatTask = sseTaskScheduler.scheduleAtFixedRate(() -> { + try { + emitter.send(SseEmitter.event().name("heartbeat").data("")); + } catch (IOException | IllegalStateException e) { + log.debug("[SSE] 心跳发送失败, userId={}, connectionId={}, reason={}", + userId, connectionId, e.getMessage()); + connectionManager.remove(userId, connectionId); + } + }, HEARTBEAT_INTERVAL); + + Runnable cancelHeartbeat = () -> heartbeatTask.cancel(false); + emitter.onCompletion(cancelHeartbeat); + emitter.onTimeout(cancelHeartbeat); + emitter.onError(error -> cancelHeartbeat.run()); - // 发送初始连接确认 try { emitter.send(SseEmitter.event() .name("connected") .data("{\"message\":\"连接成功\"}")); - } catch (IOException e) { - log.warn("[SSE] 发送 connected 事件失败, userId={}", userId); + } catch (IOException | IllegalStateException e) { + log.warn("[SSE] 发送 connected 事件失败, userId={}, connectionId={}", + userId, connectionId, e); + connectionManager.remove(userId, connectionId); } - // 启动心跳定时任务 - ScheduledExecutorService heartbeat = Executors.newSingleThreadScheduledExecutor( - r -> new Thread(r, "sse-heartbeat-" + userId.substring(0, Math.min(8, userId.length()))) - ); - heartbeat.scheduleAtFixedRate(() -> { - try { - emitter.send(SseEmitter.event() - .name("heartbeat") - .data("")); - } catch (Exception e) { - // 客户端已断开（IOException）或响应已不可用（AsyncRequestNotUsableException） - connectionManager.remove(userId); - heartbeat.shutdown(); - } - }, 30, 30, TimeUnit.SECONDS); + if (lastEventId != null && !lastEventId.isBlank()) { + log.debug("[SSE] 客户端重连, Last-Event-ID={}, userId={}, connectionId={}", + lastEventId, userId, connectionId); + } - // 连接关闭时停止心跳 - emitter.onCompletion(heartbeat::shutdown); - emitter.onTimeout(heartbeat::shutdown); - emitter.onError(e -> heartbeat.shutdown()); - - log.info("[SSE] 新连接建立, userId={}, 当前在线={}", userId, connectionManager.getOnlineCount()); + log.info("[SSE] 连接建立成功, userId={}, connectionId={}, onlineConnections={}", + userId, connectionId, connectionManager.getOnlineCount()); return emitter; } - - // ==================== 测试接口 ==================== - - /** - * 查看当前 SSE 在线连接数 - * - * GET /sse/test/online - */ - @GetMapping("/test/online") - public Map onlineCount() { - return Map.of("onlineCount", connectionManager.getOnlineCount()); - } - - /** - * 手动发送广播事件（测试用） - * - * POST /sse/test/broadcast - * Body: { "event": "room_created", "data": { "roomId": "test", ... } } - * - * event 可选值: room_created / room_updated / room_removed / kicked / game_started / balance_chg - */ - @PostMapping("/test/broadcast") - public Map testBroadcast(@RequestBody Map body) { - String eventType = (String) body.get("event"); - Object data = body.get("data"); - - if (eventType == null || data == null) { - return Map.of("success", false, "error", "event 和 data 字段为必填"); - } - - try { - SseEventType type = SseEventType.valueOf(eventType.toUpperCase()); - SseEvent event = SseEvent.broadcast(type, JSON.toJSONString(data)); - connectionManager.broadcast(event); - log.info("[SSE] 测试广播, type={}, data={}", eventType, JSON.toJSONString(data)); - return Map.of("success", true, "eventType", eventType, "onlineCount", connectionManager.getOnlineCount()); - } catch (IllegalArgumentException e) { - return Map.of("success", false, "error", "无效的事件类型: " + eventType); - } - } }