feat: sse

This commit is contained in:
2026-06-30 08:52:23 +08:00
parent aa9a0732d6
commit d92f7bb581
6 changed files with 90 additions and 14 deletions

View File

@@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSON;
import com.webgame.webgamebackend.common.exception.BusinessException; import com.webgame.webgamebackend.common.exception.BusinessException;
import com.webgame.webgamebackend.common.exception.ErrorCode; import com.webgame.webgamebackend.common.exception.ErrorCode;
import com.webgame.webgamebackend.sse.SseConnectionManager; import com.webgame.webgamebackend.sse.SseConnectionManager;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@@ -16,6 +17,7 @@ import org.springframework.stereotype.Component;
*/ */
@Slf4j @Slf4j
@Component @Component
@RequiredArgsConstructor
public class SseEventPublisher { public class SseEventPublisher {
public static final String CHANNEL = "sse:events"; public static final String CHANNEL = "sse:events";
@@ -23,12 +25,6 @@ public class SseEventPublisher {
private final RedisTemplate<String, String> redisTemplate; private final RedisTemplate<String, String> redisTemplate;
private final SseConnectionManager connectionManager; private final SseConnectionManager connectionManager;
public SseEventPublisher(RedisTemplate<String, String> redisTemplate,
SseConnectionManager connectionManager) {
this.redisTemplate = redisTemplate;
this.connectionManager = connectionManager;
}
/** /**
* 发布广播事件(推送给所有在线用户) * 发布广播事件(推送给所有在线用户)
* *

View File

@@ -9,5 +9,7 @@ public enum SseEventType {
ROOM_REMOVED, ROOM_REMOVED,
KICKED, KICKED,
GAME_STARTED, GAME_STARTED,
BALANCE_CHG BALANCE_CHG,
/** 自定义事件类型,用于测试或任意消息推送 */
CUSTOM
} }

View File

@@ -48,6 +48,21 @@ public class GlobalExceptionHandler {
// SSE 响应已被客户端关闭,不写响应体 // SSE 响应已被客户端关闭,不写响应体
} }
/**
* SSE 客户端断开连接时Spring 在 Tomcat 容器线程上抛出的 IOException
* <p>
* SSE 长连接场景下客户端主动断开(关闭标签页、网络切换等)是正常行为,
* 不应作为 ERROR 记录。仅当响应尚未提交时才视为真正的 IO 异常。
*/
@ExceptionHandler(IOException.class)
public void handleIOException(IOException ex, HttpServletResponse response) {
if (response.isCommitted()) {
log.debug("SSE 客户端已断开: {}", ex.getMessage());
} else {
log.warn("未提交响应的 IO 异常: {}", ex.getMessage());
}
}
// ==================== 业务异常REST JSON 请求) ==================== // ==================== 业务异常REST JSON 请求) ====================
/** /**
@@ -116,10 +131,25 @@ public class GlobalExceptionHandler {
// ==================== 兜底 → HTTP 500直接写响应绕过内容协商 ==================== // ==================== 兜底 → HTTP 500直接写响应绕过内容协商 ====================
@ExceptionHandler(Exception.class) @ExceptionHandler(Exception.class)
public void handleException(Exception ex, HttpServletResponse response) throws IOException { public void handleException(
Exception ex,
HttpServletResponse response
) throws IOException {
// SSE 长连接/已提交的响应 → 客户端已断开,不需要告警
if (response.isCommitted()) {
log.debug("Response 已提交,客户端已断开: {}", ex.getMessage());
return;
}
log.error("服务器内部错误", ex); log.error("服务器内部错误", ex);
writeJsonError(response, HttpStatus.INTERNAL_SERVER_ERROR,
ErrorCode.INTERNAL_ERROR.getCode(), ErrorCode.INTERNAL_ERROR.getDefaultMessage()); writeJsonError(
response,
HttpStatus.INTERNAL_SERVER_ERROR,
ErrorCode.INTERNAL_ERROR.getCode(),
ErrorCode.INTERNAL_ERROR.getDefaultMessage()
);
} }
// ==================== 工具方法 ==================== // ==================== 工具方法 ====================

View File

@@ -3,6 +3,7 @@ package com.webgame.webgamebackend.service.impl;
import com.webgame.webgamebackend.common.exception.BusinessException; import com.webgame.webgamebackend.common.exception.BusinessException;
import com.webgame.webgamebackend.common.exception.ErrorCode; import com.webgame.webgamebackend.common.exception.ErrorCode;
import com.webgame.webgamebackend.service.RefreshTokenService; import com.webgame.webgamebackend.service.RefreshTokenService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -25,6 +26,7 @@ import java.util.concurrent.TimeUnit;
*/ */
@Slf4j @Slf4j
@Service @Service
@RequiredArgsConstructor
public class RefreshTokenServiceImpl implements RefreshTokenService { public class RefreshTokenServiceImpl implements RefreshTokenService {
/** Redis key 前缀 */ /** Redis key 前缀 */
@@ -41,10 +43,6 @@ public class RefreshTokenServiceImpl implements RefreshTokenService {
private final StringRedisTemplate stringRedisTemplate; private final StringRedisTemplate stringRedisTemplate;
public RefreshTokenServiceImpl(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
@Override @Override
public String create(String userId) { public String create(String userId) {
if (userId == null || userId.isBlank()) { if (userId == null || userId.isBlank()) {

View File

@@ -1,17 +1,23 @@
package com.webgame.webgamebackend.sse; package com.webgame.webgamebackend.sse;
import com.webgame.webgamebackend.common.auth.AuthenticationProvider; import com.webgame.webgamebackend.common.auth.AuthenticationProvider;
import com.webgame.webgamebackend.common.dto.ApiResponse;
import com.webgame.webgamebackend.common.event.SseEventPublisher;
import com.webgame.webgamebackend.common.event.SseEventType;
import jakarta.servlet.http.HttpServletResponse; import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType; import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping; 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.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException; import java.io.IOException;
import java.util.Map;
/** /**
* SSE 订阅端点 * SSE 订阅端点
@@ -30,6 +36,7 @@ public class SseController {
private final SseConnectionManager connectionManager; private final SseConnectionManager connectionManager;
private final AuthenticationProvider authProvider; private final AuthenticationProvider authProvider;
private final SseEventPublisher eventPublisher;
@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE) @GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribe( public SseEmitter subscribe(
@@ -73,6 +80,49 @@ public class SseController {
return emitter; return emitter;
} }
/**
* 测试用消息发布接口
*
* 支持广播(不传 userId和定向推送传 userId两种模式。
* 可传入任意 JSON 数据作为消息体,事件类型固定为 CUSTOM。
*
* 请求体示例:
* <pre>
* // 广播
* { "data": { "msg": "hello" } }
*
* // 定向推送
* { "userId": "123", "data": { "msg": "hello" } }
* </pre>
*
* @param body 请求体,包含可选的 userId 和必填的 data
* @return 统一 API 响应
*/
@PostMapping("/publish")
public ApiResponse<Map<String, Object>> publish(@RequestBody Map<String, Object> body) {
@SuppressWarnings("unchecked")
Map<String, Object> data = (Map<String, Object>) body.get("data");
if (data == null || data.isEmpty()) {
return ApiResponse.fail(400, "data 字段不能为空");
}
String userId = body.get("userId") != null ? body.get("userId").toString() : null;
if (userId != null && !userId.isBlank()) {
eventPublisher.sendTo(userId, SseEventType.CUSTOM, data);
log.info("[SSE] 测试定向推送, userId={}, data={}", userId, data);
} else {
eventPublisher.broadcast(SseEventType.CUSTOM, data);
log.info("[SSE] 测试广播推送, data={}", data);
}
return ApiResponse.ok(Map.of(
"mode", userId != null && !userId.isBlank() ? "targeted" : "broadcast",
"userId", userId != null ? userId : "all",
"data", data
));
}
/** /**
* 向 SSE 客户端返回认证错误 * 向 SSE 客户端返回认证错误
* *

Binary file not shown.

After

Width:  |  Height:  |  Size: 16 KiB