fix: 暂存

src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java

@@ -1,7 +1,9 @@
 package com.webgame.webgamebackend.common.event;
 
 import com.alibaba.fastjson2.JSON;
-import lombok.RequiredArgsConstructor;
+import com.webgame.webgamebackend.common.exception.BusinessException;
+import com.webgame.webgamebackend.common.exception.ErrorCode;
+import com.webgame.webgamebackend.sse.SseConnectionManager;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
@@ -9,34 +11,80 @@ import org.springframework.stereotype.Component;
 /**
  * 统一 SSE 事件发布器
  *
- * 业务代码通过此发布器发布领域无关事件，由 Redis Pub/Sub 扇出到所有应用实例。
+ * 优先通过 Redis Pub/Sub 扇出事件到所有应用实例；
+ * Redis 不可用时降级为本地直推，并抛出异常让调用方事务回滚。
  */
 @Slf4j
 @Component
-@RequiredArgsConstructor
 public class SseEventPublisher {
 
     public static final String CHANNEL = "sse:events";
 
     private final RedisTemplate<String, String> redisTemplate;
+    private final SseConnectionManager connectionManager;
 
+    public SseEventPublisher(RedisTemplate<String, String> redisTemplate,
+                             SseConnectionManager connectionManager) {
+        this.redisTemplate = redisTemplate;
+        this.connectionManager = connectionManager;
+    }
+
+    /**
+     * 发布广播事件（推送给所有在线用户）
+     *
+     * @param type 事件类型
+     * @param data 事件数据对象（将被序列化为 JSON）
+     * @throws BusinessException Redis 和本地降级均失败时抛出
+     */
     public void broadcast(SseEventType type, Object data) {
         publish(SseEvent.broadcast(type, JSON.toJSONString(data)));
     }
 
+    /**
+     * 发布定向事件（只推送给指定用户）
+     *
+     * @param userId 目标用户账号 ID
+     * @param type   事件类型
+     * @param data   事件数据对象（将被序列化为 JSON）
+     * @throws BusinessException Redis 和本地降级均失败时抛出
+     */
     public void sendTo(String userId, SseEventType type, Object data) {
         publish(SseEvent.targeted(type, userId, JSON.toJSONString(data)));
     }
 
+    /**
+     * 发布事件到 Redis Pub/Sub 频道
+     *
+     * 优先走 Redis 扇出到所有实例；Redis 不可用时降级为直接推送给本地连接。
+     * 两次尝试均失败则抛出 BusinessException，让调用方事务回滚。
+     *
+     * @param event 事件对象
+     * @throws BusinessException 发布完全失败时抛出
+     */
     private void publish(SseEvent event) {
+        // 优先：通过 Redis Pub/Sub 扇出
         try {
             String json = JSON.toJSONString(event);
             redisTemplate.convertAndSend(CHANNEL, json);
             log.debug("[SSE] 事件已发布, channel={}, type={}, targetId={}",
                     CHANNEL, event.type(), event.targetId());
-        } catch (Exception e) {
+            return;
-            log.error("[SSE] 发布事件失败, type={}, targetId={}",
+        } catch (Exception redisEx) {
-                    event.type(), event.targetId(), e);
+            log.warn("[SSE] Redis 不可用，降级为本地直推, type={}, targetId={}, reason={}",
+                    event.type(), event.targetId(), redisEx.getMessage());
+        }
+
+        // 降级：直接推送给本地连接（本实例用户不丢事件）
+        try {
+            if (event.targetId() == null) {
+                connectionManager.broadcast(event);
+            } else {
+                connectionManager.sendTo(event.targetId(), event);
+            }
+            log.info("[SSE] 已降级为本地直推, type={}, targetId={}", event.type(), event.targetId());
+        } catch (Exception localEx) {
+            log.error("[SSE] 本地直推也失败, type={}, targetId={}", event.type(), event.targetId(), localEx);
+            throw new BusinessException(ErrorCode.INTERNAL_ERROR, "事件推送失败，请重试");
         }
     }
 }

src/main/java/com/webgame/webgamebackend/config/SseConfig.java

@@ -7,11 +7,12 @@ 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 基础设施配置
+ *
+ * 注册 Redis Pub/Sub 消息监听容器，订阅频道 "sse:events"。
+ * 心跳由 SseConnectionManager 自管理单线程调度，不在此处配置。
  */
 @Configuration
 public class SseConfig {
@@ -25,14 +26,4 @@ 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;
-    }
 }

src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java

@@ -1,6 +1,8 @@
 package com.webgame.webgamebackend.sse;
 
 import com.webgame.webgamebackend.common.event.SseEvent;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
@@ -9,42 +11,99 @@ import java.io.IOException;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 线程安全的 SSE 连接管理器
  *
  * 一个用户可以拥有多个连接，使得不同浏览器标签页和设备之间不会互相踢下线。
+ * 心跳由单一后台线程统一管理，避免 per-connection 定时器导致的线程膨胀。
  */
 @Slf4j
 @Component
 public class SseConnectionManager {
 
+    /** 心跳间隔（秒） */
+    private static final long HEARTBEAT_INTERVAL_SECONDS = 30;
+
+    /** 心跳线程池（单线程，批量 ping 所有连接） */
+    private final ScheduledExecutorService heartbeatScheduler =
+            Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "sse-heartbeat"));
+
     private final ConcurrentHashMap<String, ConcurrentHashMap<String, SseEmitter>> emitters =
             new ConcurrentHashMap<>();
 
+    /**
+     * 启动统一心跳定时任务
+     *
+     * 单一后台线程批量 ping 所有连接，O(1) 线程开销。
+     * 单次 ping 失败自动清理连接，不影响其他连接。
+     */
+    @PostConstruct
+    public void startHeartbeat() {
+        heartbeatScheduler.scheduleAtFixedRate(() -> {
+            if (emitters.isEmpty()) {
+                return;
+            }
+            log.debug("[SSE] 批量心跳, onlineConnections={}", getOnlineCount());
+            emitters.forEach((userId, connections) ->
+                    connections.forEach((connectionId, emitter) -> {
+                        try {
+                            emitter.send(SseEmitter.event().name("heartbeat").data(""));
+                        } catch (IOException | IllegalStateException e) {
+                            log.debug("[SSE] 心跳失败，清理连接, userId={}, connectionId={}",
+                                    userId, connectionId);
+                            remove(userId, connectionId);
+                        }
+                    }));
+        }, HEARTBEAT_INTERVAL_SECONDS, HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS);
+        log.info("[SSE] 心跳定时器已启动, interval={}s", HEARTBEAT_INTERVAL_SECONDS);
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        heartbeatScheduler.shutdown();
+        log.info("[SSE] 心跳定时器已关闭");
+    }
+
+    /**
+     * 注册新的 SSE 连接
+     *
+     * @param userId  用户账号 ID
+     * @param emitter SseEmitter 实例
+     * @return 连接 ID
+     */
     public String register(String userId, SseEmitter emitter) {
         String connectionId = UUID.randomUUID().toString();
         emitters.computeIfAbsent(userId, key -> new ConcurrentHashMap<>()).put(connectionId, emitter);
 
         emitter.onCompletion(() -> {
-            log.debug("[SSE] Connection completed, userId={}, connectionId={}", userId, connectionId);
+            log.debug("[SSE] 连接完成, userId={}, connectionId={}", userId, connectionId);
             remove(userId, connectionId);
         });
         emitter.onTimeout(() -> {
-            log.debug("[SSE] Connection timeout, userId={}, connectionId={}", userId, connectionId);
+            log.debug("[SSE] 连接超时, userId={}, connectionId={}", userId, connectionId);
             remove(userId, connectionId);
         });
         emitter.onError(throwable -> {
-            log.debug("[SSE] Connection error, userId={}, connectionId={}, reason={}",
+            log.debug("[SSE] 连接异常, userId={}, connectionId={}, reason={}",
                     userId, connectionId, throwable.getMessage());
             remove(userId, connectionId);
         });
 
-        log.info("[SSE] User connected, userId={}, connectionId={}, onlineConnections={}",
+        log.info("[SSE] 用户连接成功, userId={}, connectionId={}, onlineConnections={}",
                 userId, connectionId, getOnlineCount());
         return connectionId;
     }
 
+    /**
+     * 移除连接
+     *
+     * @param userId       用户账号 ID
+     * @param connectionId 连接 ID
+     */
     public void remove(String userId, String connectionId) {
         ConcurrentHashMap<String, SseEmitter> userEmitters = emitters.get(userId);
         if (userEmitters == null) {
@@ -62,35 +121,56 @@ public class SseConnectionManager {
             } catch (Throwable ignored) {
                 // 客户端断开后响应可能已不可用
             }
-            log.info("[SSE] Connection removed, userId={}, connectionId={}, onlineConnections={}",
+            log.info("[SSE] 连接已移除, userId={}, connectionId={}, onlineConnections={}",
                     userId, connectionId, getOnlineCount());
         }
     }
 
+    /**
+     * 广播事件给所有在线用户的所有连接
+     *
+     * @param event SSE 事件
+     */
     public void broadcast(SseEvent event) {
         if (emitters.isEmpty()) {
             return;
         }
 
-        log.debug("[SSE] Broadcast event, type={}, onlineConnections={}", event.type(), getOnlineCount());
+        log.debug("[SSE] 广播事件, type={}, onlineConnections={}", event.type(), getOnlineCount());
         emitters.forEach((userId, connections) ->
                 connections.forEach((connectionId, emitter) -> sendEvent(userId, connectionId, emitter, event)));
     }
 
+    /**
+     * 发送事件给指定用户的所有连接
+     *
+     * @param userId 目标用户账号 ID
+     * @param event  SSE 事件
+     */
     public void sendTo(String userId, SseEvent event) {
         ConcurrentHashMap<String, SseEmitter> connections = emitters.get(userId);
         if (connections == null || connections.isEmpty()) {
-            log.debug("[SSE] Target user is offline, userId={}, eventType={}", userId, event.type());
+            log.debug("[SSE] 目标用户不在线, userId={}, eventType={}", userId, event.type());
             return;
         }
 
         connections.forEach((connectionId, emitter) -> sendEvent(userId, connectionId, emitter, event));
     }
 
+    /**
+     * 获取当前在线连接总数
+     *
+     * @return 连接数
+     */
     public int getOnlineCount() {
         return emitters.values().stream().mapToInt(Map::size).sum();
     }
 
+    /**
+     * 向单个 SseEmitter 发送事件
+     *
+     * 捕获 IOException（客户端已断开但还未触发回调），此时安全移除连接。
+     */
     private void sendEvent(String userId, String connectionId, SseEmitter emitter, SseEvent event) {
         try {
             emitter.send(SseEmitter.event()
@@ -98,7 +178,7 @@ public class SseConnectionManager {
                     .name(event.type().name().toLowerCase())
                     .data(event.data()));
         } catch (IOException | IllegalStateException e) {
-            log.debug("[SSE] Push failed, userId={}, connectionId={}, reason={}",
+            log.debug("[SSE] 推送失败, userId={}, connectionId={}, reason={}",
                     userId, connectionId, e.getMessage());
             remove(userId, connectionId);
         }

src/main/java/com/webgame/webgamebackend/sse/SseController.java

@@ -4,7 +4,6 @@ import cn.dev33.satoken.stp.StpUtil;
 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.RequestHeader;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -12,11 +11,12 @@ import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.io.IOException;
-import java.time.Duration;
-import java.util.concurrent.ScheduledFuture;
 
 /**
  * SSE 订阅端点
+ *
+ * 提供 GET /sse/subscribe 端点建立持久化 SSE 连接。
+ * 心跳由 SseConnectionManager 统一批量管理，此处不再维护 per-connection 定时器。
  */
 @Slf4j
 @RestController
@@ -24,11 +24,10 @@ import java.util.concurrent.ScheduledFuture;
 @RequiredArgsConstructor
 public class SseController {
 
+    /** SseEmitter 超时时间（30 分钟），超时后由 SseConnectionManager 的 onTimeout 回调清理 */
     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;
 
     @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     public SseEmitter subscribe(
@@ -37,21 +36,7 @@ public class SseController {
         SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);
         String connectionId = 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")

src/main/java/com/webgame/webgamebackend/ws/RoomWebSocketHandler.java

@@ -7,6 +7,7 @@ import com.webgame.webgamebackend.service.GomokuGameService;
 import com.webgame.webgamebackend.ws.dto.WsInboundMessage;
 import com.webgame.webgamebackend.ws.dto.WsMessage;
 import com.webgame.webgamebackend.ws.dto.WsOutboundMessage;
+import jakarta.annotation.PostConstruct;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -22,7 +23,8 @@ import java.util.Map;
 /**
  * 房间 WebSocket 处理器
  *
- * 处理客户端连接、断开与消息路由。消息类型路由到不同的业务逻辑。
+ * 处理客户端连接、断开与消息路由。消息类型通过 WsMessageRouter 注册表分发，
+ * 新增消息类型只需注册对应的 WsMessageHandler，无需修改路由逻辑。
  * 通过 URL 参数传递 roomId 和 token 完成认证与房间绑定。
  */
 @Slf4j
@@ -32,6 +34,56 @@ public class RoomWebSocketHandler extends TextWebSocketHandler {
 
     private final RoomSessionManager sessionManager;
     private final GomokuGameService gomokuGameService;
+    private final WsMessageRouter router;
+
+    /**
+     * 注册所有消息类型与对应处理器
+     *
+     * 新增消息类型只需在此方法中添加一行 router.register(...) 即可。
+     */
+    @PostConstruct
+    public void registerHandlers() {
+        // 落子：需从 data 中提取 row, col
+        router.register(WsInboundMessage.PLACE_STONE, (roomId, userId, data, sm) -> {
+            int row = data.getIntValue("row");
+            int col = data.getIntValue("col");
+            gomokuGameService.placeStone(roomId, userId, row, col, sm);
+        });
+
+        // 聊天
+        router.register(WsInboundMessage.CHAT, (roomId, userId, data, sm) -> {
+            String content = data.getString("content");
+            handleChat(roomId, userId, content);
+        });
+
+        // 准备/取消准备
+        router.register(WsInboundMessage.READY,
+                (roomId, userId, data, sm) -> gomokuGameService.toggleReady(roomId, userId, sm));
+
+        // 认输
+        router.register(WsInboundMessage.RESIGN,
+                (roomId, userId, data, sm) -> gomokuGameService.resign(roomId, userId, sm));
+
+        // 求和请求
+        router.register(WsInboundMessage.DRAW_REQUEST,
+                (roomId, userId, data, sm) -> gomokuGameService.requestDraw(roomId, userId, sm));
+
+        // 回应求和：需从 data 中提取 accept
+        router.register(WsInboundMessage.DRAW_RESPONSE, (roomId, userId, data, sm) -> {
+            boolean accept = data.getBooleanValue("accept");
+            gomokuGameService.respondToDraw(roomId, userId, accept, sm);
+        });
+
+        // 离开房间
+        router.register(WsInboundMessage.LEAVE,
+                (roomId, userId, data, sm) -> gomokuGameService.leaveRoom(roomId, userId, sm));
+
+        // 开始游戏（仅房主）
+        router.register(WsInboundMessage.START,
+                (roomId, userId, data, sm) -> gomokuGameService.startGame(roomId, userId, sm));
+
+        log.info("[WS] 已注册 {} 个消息处理器", router.size());
+    }
 
     @Override
     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
@@ -89,40 +141,17 @@ public class RoomWebSocketHandler extends TextWebSocketHandler {
         try {
             JSONObject json = JSON.parseObject(payload);
             String type = json.getString("type");
-            Object data = json.get("data");
+            Object rawData = json.get("data");
+            JSONObject data = (rawData instanceof JSONObject) ? (JSONObject) rawData : null;
 
-            switch (type) {
+            // 通过路由器查找处理器
-                case WsInboundMessage.PLACE_STONE -> {
+            WsMessageHandler handler = router.get(type);
-                    // data 包含 row, col
+            if (handler == null) {
-                    int row = ((JSONObject) data).getIntValue("row");
+                sendError(session, "未知消息类型: " + type);
-                    int col = ((JSONObject) data).getIntValue("col");
+                return;
-                    gomokuGameService.placeStone(roomId, userId, row, col, sessionManager);
-                }
-                case WsInboundMessage.CHAT -> {
-                    String content = ((JSONObject) data).getString("content");
-                    handleChat(roomId, userId, content);
-                }
-                case WsInboundMessage.READY -> {
-                    gomokuGameService.toggleReady(roomId, userId, sessionManager);
-                }
-                case WsInboundMessage.RESIGN -> {
-                    gomokuGameService.resign(roomId, userId, sessionManager);
-                }
-                case WsInboundMessage.DRAW_REQUEST -> {
-                    gomokuGameService.requestDraw(roomId, userId, sessionManager);
-                }
-                case WsInboundMessage.DRAW_RESPONSE -> {
-                    boolean accept = ((JSONObject) data).getBooleanValue("accept");
-                    gomokuGameService.respondToDraw(roomId, userId, accept, sessionManager);
-                }
-                case WsInboundMessage.LEAVE -> {
-                    gomokuGameService.leaveRoom(roomId, userId, sessionManager);
-                }
-                case WsInboundMessage.START -> {
-                    gomokuGameService.startGame(roomId, userId, sessionManager);
-                }
-                default -> sendError(session, "未知消息类型: " + type);
             }
+
+            handler.handle(roomId, userId, data, sessionManager);
         } catch (Exception e) {
             log.error("[WS] 处理消息异常, roomId={}, userId={}", roomId, userId, e);
             sendError(session, e.getMessage());

src/main/java/com/webgame/webgamebackend/ws/WsMessageHandler.java

@@ -0,0 +1,23 @@
+package com.webgame.webgamebackend.ws;
+
+import com.alibaba.fastjson2.JSONObject;
+
+/**
+ * WebSocket 消息处理器接口
+ *
+ * 每种消息类型对应一个处理器实现，由 WsMessageRouter 统一路由。
+ * handler 方法抛出的异常由 RoomWebSocketHandler 统一捕获并返回错误消息。
+ */
+@FunctionalInterface
+public interface WsMessageHandler {
+
+    /**
+     * 处理 WebSocket 消息
+     *
+     * @param roomId         房间 ID
+     * @param userId         发送消息的用户 ID
+     * @param data           消息数据（JSONObject，可能为 null）
+     * @param sessionManager 房间会话管理器，用于广播
+     */
+    void handle(String roomId, String userId, JSONObject data, RoomSessionManager sessionManager);
+}

src/main/java/com/webgame/webgamebackend/ws/WsMessageRouter.java

@@ -0,0 +1,72 @@
+package com.webgame.webgamebackend.ws;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * WebSocket 消息路由器
+ *
+ * 维护消息类型 → 处理器的映射表，替代硬编码 switch-case。
+ * 新增消息类型只需注册对应的 WsMessageHandler 即可，无需修改路由逻辑。
+ *
+ * 使用示例：
+ * <pre>{@code
+ * router.register("place_stone", gomokuGameService::placeStoneViaWs);
+ * router.register("chat", chatHandler::handle);
+ * }</pre>
+ */
+@Slf4j
+@Component
+public class WsMessageRouter {
+
+    private final Map<String, WsMessageHandler> handlers = new ConcurrentHashMap<>();
+
+    /**
+     * 注册消息处理器
+     *
+     * @param type    消息类型（与客户端约定的一致）
+     * @param handler 处理器
+     */
+    public void register(String type, WsMessageHandler handler) {
+        if (type == null || type.isBlank()) {
+            throw new IllegalArgumentException("消息类型不能为空");
+        }
+        if (handler == null) {
+            throw new IllegalArgumentException("处理器不能为 null");
+        }
+        handlers.put(type, handler);
+        log.info("[WS路由] 注册处理器, type={}", type);
+    }
+
+    /**
+     * 获取消息处理器
+     *
+     * @param type 消息类型
+     * @return 处理器，未找到返回 null
+     */
+    public WsMessageHandler get(String type) {
+        return handlers.get(type);
+    }
+
+    /**
+     * 是否已注册指定类型的处理器
+     *
+     * @param type 消息类型
+     * @return 是否已注册
+     */
+    public boolean hasHandler(String type) {
+        return handlers.containsKey(type);
+    }
+
+    /**
+     * 获取已注册的消息类型数量（用于监控/调试）
+     *
+     * @return 注册数量
+     */
+    public int size() {
+        return handlers.size();
+    }
+}