feat: 添加 SseController SSE 订阅端点
This commit is contained in:
@@ -0,0 +1,79 @@
|
|||||||
|
package com.webgame.webgamebackend.sse;
|
||||||
|
|
||||||
|
import cn.dev33.satoken.stp.StpUtil;
|
||||||
|
import lombok.RequiredArgsConstructor;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.springframework.http.MediaType;
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
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.concurrent.Executors;
|
||||||
|
import java.util.concurrent.ScheduledExecutorService;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* SSE 连接端点
|
||||||
|
*
|
||||||
|
* 提供 GET /sse/subscribe 端点,建立持久化 SSE 连接。
|
||||||
|
* 连接成功后立即发送 connected 事件确认握手,之后每 30 秒发送心跳保活。
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("/sse")
|
||||||
|
@RequiredArgsConstructor
|
||||||
|
public class SseController {
|
||||||
|
|
||||||
|
private final SseConnectionManager connectionManager;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 建立 SSE 订阅连接
|
||||||
|
*
|
||||||
|
* 需要登录鉴权(Sa-Token 自动校验)。
|
||||||
|
* EventSource API 不支持自定义请求头,token 通过 URL 查询参数 ?token=xxx 传递。
|
||||||
|
*
|
||||||
|
* @return SseEmitter 实例(30 分钟超时)
|
||||||
|
*/
|
||||||
|
@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
|
||||||
|
public SseEmitter subscribe() {
|
||||||
|
String userId = StpUtil.getLoginIdAsString();
|
||||||
|
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30 分钟超时
|
||||||
|
|
||||||
|
connectionManager.register(userId, emitter);
|
||||||
|
|
||||||
|
// 发送初始连接确认
|
||||||
|
try {
|
||||||
|
emitter.send(SseEmitter.event()
|
||||||
|
.name("connected")
|
||||||
|
.data("{\"message\":\"连接成功\"}"));
|
||||||
|
} catch (IOException e) {
|
||||||
|
log.warn("[SSE] 发送 connected 事件失败, userId={}", userId);
|
||||||
|
}
|
||||||
|
|
||||||
|
// 启动心跳定时任务
|
||||||
|
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 (IOException e) {
|
||||||
|
// 客户端已断开,清理
|
||||||
|
connectionManager.remove(userId);
|
||||||
|
heartbeat.shutdown();
|
||||||
|
}
|
||||||
|
}, 30, 30, TimeUnit.SECONDS);
|
||||||
|
|
||||||
|
// 连接关闭时停止心跳
|
||||||
|
emitter.onCompletion(heartbeat::shutdown);
|
||||||
|
emitter.onTimeout(heartbeat::shutdown);
|
||||||
|
emitter.onError(e -> heartbeat.shutdown());
|
||||||
|
|
||||||
|
log.info("[SSE] 新连接建立, userId={}, 当前在线={}", userId, connectionManager.getOnlineCount());
|
||||||
|
return emitter;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user