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 a18b0a2..f9cf518 100644 --- a/src/main/java/com/webgame/webgamebackend/common/event/SseEventPublisher.java +++ b/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 redisTemplate; + private final SseConnectionManager connectionManager; + public SseEventPublisher(RedisTemplate 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) { - log.error("[SSE] 发布事件失败, type={}, targetId={}", - event.type(), event.targetId(), e); + return; + } catch (Exception redisEx) { + 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, "事件推送失败，请重试"); } } } diff --git a/src/main/java/com/webgame/webgamebackend/config/SseConfig.java b/src/main/java/com/webgame/webgamebackend/config/SseConfig.java index 2486b22..7cc1c9c 100644 --- a/src/main/java/com/webgame/webgamebackend/config/SseConfig.java +++ b/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; - } } diff --git a/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java b/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java index fd61cdf..19f2da8 100644 --- a/src/main/java/com/webgame/webgamebackend/sse/SseConnectionManager.java +++ b/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> 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 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 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); } diff --git a/src/main/java/com/webgame/webgamebackend/sse/SseController.java b/src/main/java/com/webgame/webgamebackend/sse/SseController.java index c87ad69..55c5748 100644 --- a/src/main/java/com/webgame/webgamebackend/sse/SseController.java +++ b/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") diff --git a/src/main/java/com/webgame/webgamebackend/ws/RoomWebSocketHandler.java b/src/main/java/com/webgame/webgamebackend/ws/RoomWebSocketHandler.java index c6406f1..7087ae5 100644 --- a/src/main/java/com/webgame/webgamebackend/ws/RoomWebSocketHandler.java +++ b/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 -> { - // data 包含 row, col - int row = ((JSONObject) data).getIntValue("row"); - int col = ((JSONObject) data).getIntValue("col"); - 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); + // 通过路由器查找处理器 + WsMessageHandler handler = router.get(type); + if (handler == null) { + sendError(session, "未知消息类型: " + type); + return; } + + handler.handle(roomId, userId, data, sessionManager); } catch (Exception e) { log.error("[WS] 处理消息异常, roomId={}, userId={}", roomId, userId, e); sendError(session, e.getMessage()); diff --git a/src/main/java/com/webgame/webgamebackend/ws/WsMessageHandler.java b/src/main/java/com/webgame/webgamebackend/ws/WsMessageHandler.java new file mode 100644 index 0000000..2d7efd1 --- /dev/null +++ b/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); +} diff --git a/src/main/java/com/webgame/webgamebackend/ws/WsMessageRouter.java b/src/main/java/com/webgame/webgamebackend/ws/WsMessageRouter.java new file mode 100644 index 0000000..ad625f1 --- /dev/null +++ b/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 即可，无需修改路由逻辑。 + * + * 使用示例： + * 
{@code
+ * router.register("place_stone", gomokuGameService::placeStoneViaWs);
+ * router.register("chat", chatHandler::handle);
+ * }
+ */ +@Slf4j +@Component +public class WsMessageRouter { + + private final Map 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(); + } +}