Skip to content

G
gk
Commit 10f3fd13 authored by daniel's avatar daniel
Browse files
Options

fix(websocket): websocket接口调整

parent beb3de1a
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 201 additions and 386 deletions
src/main/java/com/dlbridge/jadp/config/web/SchedulerConfig.java 0 → 100644
View file @ 10f3fd13
package com.dlbridge.jadp.config.web;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@Configuration
@EnableScheduling
public class SchedulerConfig {
@Bean
public ThreadPoolTaskScheduler taskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(2);
scheduler.setThreadNamePrefix("scheduled-task-");
return scheduler;
}
}
src/main/java/com/dlbridge/jadp/config/web/WebSocketConfig.java
View file @ 10f3fd13
package com.dlbridge.jadp.config.web; package com.dlbridge.jadp.config.web;
import com.dlbridge.jadp.websocket.TicketWebSocketHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; import org.springframework.web.socket.config.annotation.*;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
@Configuration @Configuration
@EnableWebSocketMessageBroker @EnableWebSocket
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { public class WebSocketConfig implements WebSocketConfigurer {
@Override private final TicketWebSocketHandler handler;
public void configureMessageBroker(MessageBrokerRegistry config) {
// 创建一个调度器，用于心跳定时任务 public WebSocketConfig(TicketWebSocketHandler handler) {
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); this.handler = handler;
taskScheduler.setPoolSize(1);
taskScheduler.setThreadNamePrefix("wss-heartbeat-thread-");
taskScheduler.initialize();
// 启用简单消息代理，并配置心跳
config.enableSimpleBroker("/topic", "/queue")
.setHeartbeatValue(new long[]{10000, 10000}) // 服务端和客户端心跳间隔（毫秒）
.setTaskScheduler(taskScheduler);
// 设置应用程序目的地前缀
config.setApplicationDestinationPrefixes("/app");
// 设置用户目的地前缀（点对点消息）
config.setUserDestinationPrefix("/user");
} }
@Override @Override
public void registerStompEndpoints(StompEndpointRegistry registry) { public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
// 注册 STOMP 端点，指定使用 SockJS registry.addHandler(handler, "/ws/ticket")
registry.addEndpoint("/ws").setAllowedOrigins("*") // 允许所有来源，生产环境应指定具体域名 .setAllowedOrigins("*"); // 原生 WebSocket, 不使用 SockJS
.withSockJS() }
.setHeartbeatTime(10000);
// 同时支持原生 WebSocket // ✅ 这里就是声明 ThreadPoolTaskScheduler Bean
registry.addEndpoint("/ws") @Bean
.setAllowedOrigins("*"); public ThreadPoolTaskScheduler messageBrokerTaskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(1);
scheduler.setThreadNamePrefix("wss-heartbeat-thread-");
return scheduler;
} }
} }
\ No newline at end of file
src/main/java/com/dlbridge/jadp/websocket/TicketWebSocketHandler.java 0 → 100644
View file @ 10f3fd13
package com.dlbridge.jadp.websocket;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.util.Map;
@Component
public class TicketWebSocketHandler extends TextWebSocketHandler {
private final WebSocketSessionService sessionService;
public TicketWebSocketHandler(WebSocketSessionService sessionService) {
this.sessionService = sessionService;
}
@Override
public void afterConnectionEstablished(WebSocketSession session) {
System.out.println("新连接: " + session.getId());
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
JSONObject msg = JSONUtil.parseObj(message.getPayload());
String action = msg.getStr("action");
String ticketNo = msg.getStr("ticketNo");
WebSocketSessionWrapper wrapper = new WebSocketSessionWrapper(session);
switch (action) {
case "subscribe":
sessionService.registerUserSession(ticketNo, ticketNo, session.getId(), wrapper);
wrapper.sendMessage(WebSocketMessage.info(ticketNo, "订阅成功").toJson());
break;
case "unsubscribe":
sessionService.removeTicketSession(session.getId(), ticketNo);
wrapper.sendMessage(WebSocketMessage.info(ticketNo, "取消订阅成功").toJson());
break;
case "heartbeat":
// 可记录心跳时间
break;
}
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
sessionService.removeUserSession(session.getId());
System.out.println("连接关闭: " + session.getId());
}
}
\ No newline at end of file
src/main/java/com/dlbridge/jadp/websocket/WebSocketController.java deleted 100644 → 0
View file @ beb3de1a
package com.dlbridge.jadp.websocket;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.simp.annotation.SubscribeMapping;
import org.springframework.stereotype.Controller;
import java.util.Map;
@Slf4j
@Controller
@RequiredArgsConstructor
public class WebSocketController {
private final WebSocketSessionService sessionService;
/**
* 处理客户端发送的心跳消息
*/
@MessageMapping("/heartbeat")
public void handleHeartbeat(@Payload Map<String, Object> message) {
String sessionId = (String) message.get("sessionId");
String userId = (String) message.get("userId");
log.info("收到心跳消息 - 会话ID: {}, 用户ID: {}", sessionId, userId);
}
/**
* 处理客户端订阅特定票号
*/
@MessageMapping("/ticket/subscribe")
public void handleTicketSubscribe(@Payload Map<String, Object> message, SimpMessageHeaderAccessor headerAccessor) {
// String sessionId = (String) message.get("sessionId");
String sessionId = headerAccessor.getSessionId();
String userId = (String) message.get("userId");
String ticketNo = (String) message.get("ticketNo");
log.info("客户端订阅票号 - 会话ID: {}, 用户ID: {}, 票号: {}", sessionId, userId, ticketNo);
if (userId != null && ticketNo != null) {
sessionService.registerUserSession(userId, ticketNo, sessionId);
// 发送订阅成功确认
WebSocketMessage response = WebSocketMessage.info(
ticketNo,
"票号订阅成功，开始接收核销通知"
);
sessionService.sendToUser(userId, response);
}
}
/**
* 处理客户端取消订阅票号
*/
@MessageMapping("/ticket/unsubscribe")
public void handleTicketUnsubscribe(@Payload Map<String, Object> message) {
String sessionId = (String) message.get("sessionId");
String ticketNo = (String) message.get("ticketNo");
log.info("客户端取消订阅票号 - 会话ID: {}, 票号: {}", sessionId, ticketNo);
sessionService.removeTicketSession(sessionId, ticketNo);
}
/**
* 处理获取会话状态请求
*/
@SubscribeMapping("/user/queue/status")
public Map<String, Object> handleStatusRequest() {
return sessionService.getSessionStats();
}
}
\ No newline at end of file
src/main/java/com/dlbridge/jadp/websocket/WebSocketEventListener.java deleted 100644 → 0
View file @ beb3de1a
package com.dlbridge.jadp.websocket;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.messaging.SessionConnectEvent;
import org.springframework.web.socket.messaging.SessionDisconnectEvent;
import org.springframework.web.socket.messaging.SessionSubscribeEvent;
@Slf4j
@Component
@RequiredArgsConstructor
public class WebSocketEventListener {
private final WebSocketSessionService sessionService;
/**
* 处理 WebSocket 连接建立事件
*/
@EventListener
public void handleWebSocketConnectListener(SessionConnectEvent event) {
StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
String sessionId = headerAccessor.getSessionId();
String userId = headerAccessor.getFirstNativeHeader("userId");
log.info("WebSocket 连接建立 - 会话ID: {}, 用户ID: {}", sessionId, userId);
}
/**
* 处理 WebSocket 订阅事件
*/
@EventListener
public void handleWebSocketSubscribeListener(SessionSubscribeEvent event) {
StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
String sessionId = headerAccessor.getSessionId();
String destination = headerAccessor.getDestination();
String userId = headerAccessor.getFirstNativeHeader("userId");
String ticketNo = headerAccessor.getFirstNativeHeader("ticketNo");
log.info("WebSocket 订阅 - 会话ID: {}, 目的地: {}, 用户ID: {}, 票号: {}",
sessionId, destination, userId, ticketNo);
if (userId != null && ticketNo != null) {
sessionService.registerUserSession(userId, ticketNo, sessionId);
}
}
/**
* 处理 WebSocket 连接断开事件
*/
@EventListener
public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
String sessionId = headerAccessor.getSessionId();
log.info("WebSocket 连接断开 - 会话ID: {}", sessionId);
sessionService.removeUserSession(sessionId);
}
}
src/main/java/com/dlbridge/jadp/websocket/WebSocketMessage.java
View file @ 10f3fd13
package com.dlbridge.jadp.websocket; package com.dlbridge.jadp.websocket;
import cn.hutool.json.JSONObject;
import lombok.AllArgsConstructor; import cn.hutool.json.JSONUtil;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor; import lombok.AllArgsConstructor;
import java.time.LocalDateTime;
@Data @Data
@NoArgsConstructor
@AllArgsConstructor @AllArgsConstructor
public class WebSocketMessage { public class WebSocketMessage {
private String type; // 消息类型：VERIFICATION_SUCCESS, ERROR, etc.
private String ticketNo; // 票号
private String message; // 消息内容
private LocalDateTime timestamp; // 时间戳
private Object data; // 附加数据
private String type; // 消息类型 success / error / info
private String ticketNo; // 票号
private String message; // 消息内容
private Object data; // 附加数据
// 定义一个内部的枚举类型，用于表示消息类型，包括成功、错误和信息三种类型，里面存放对应的字符串值
public enum MessageType {
SUCCESS("success"), ERROR("error"), INFO("info");
private final String value;
MessageType(String value) {
this.value = value;
}
public String getValue() {
return value;
}
}
// 构建消息静态方法
public static WebSocketMessage success(String ticketNo, String message) { public static WebSocketMessage success(String ticketNo, String message) {
return new WebSocketMessage("VERIFICATION_SUCCESS", ticketNo, message, LocalDateTime.now(), null); return new WebSocketMessage(MessageType.SUCCESS.getValue(), ticketNo, message, null);
}
public static WebSocketMessage success(String ticketNo, String message, Object data) {
return new WebSocketMessage(MessageType.SUCCESS.getValue(), ticketNo, message, data);
} }
public static WebSocketMessage error(String ticketNo, String message) { public static WebSocketMessage error(String ticketNo, String message) {
return new WebSocketMessage("ERROR", ticketNo, message, LocalDateTime.now(), null); return new WebSocketMessage(MessageType.ERROR.getValue(), ticketNo, message, null);
}
public static WebSocketMessage error(String ticketNo, String message, Object data) {
return new WebSocketMessage(MessageType.ERROR.getValue(), ticketNo, message, data);
} }
public static WebSocketMessage info(String ticketNo, String message) { public static WebSocketMessage info(String ticketNo, String message) {
return new WebSocketMessage("INFO", ticketNo, message, LocalDateTime.now(), null); return new WebSocketMessage(MessageType.INFO.getValue(), ticketNo, message, null);
}
public static WebSocketMessage info(String ticketNo, String message, Object data) {
return new WebSocketMessage(MessageType.INFO.getValue(), ticketNo, message, data);
}
// 转 JSON 字符串
public String toJson() {
JSONObject obj = JSONUtil.createObj()
.set("type", type)
.set("ticketNo", ticketNo)
.set("message", message);
if (data != null) {
obj.set("data", data);
}
return obj.toString();
} }
} }
\ No newline at end of file
src/main/java/com/dlbridge/jadp/websocket/WebSocketSessionWrapper.java 0 → 100644
View file @ 10f3fd13
package com.dlbridge.jadp.websocket;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
public class WebSocketSessionWrapper {
private final WebSocketSession session;
public WebSocketSessionWrapper(WebSocketSession session) {
this.session = session;
}
public void sendMessage(String payload) {
try {
if (session.isOpen()) session.sendMessage(new TextMessage(payload));
} catch (Exception e) {
e.printStackTrace();
}
}
public String getId() {
return session.getId();
}
}
\ No newline at end of file
src/main/java/com/dlbridge/jadp/websocket/WebSocketTestController.java
View file @ 10f3fd13
...@@ -15,22 +15,6 @@ public class WebSocketTestController { ...@@ -15,22 +15,6 @@ public class WebSocketTestController {
public WebSocketTestController(WebSocketSessionService sessionService) { public WebSocketTestController(WebSocketSessionService sessionService) {
this.sessionService = sessionService; this.sessionService = sessionService;
} }
/**
* 测试接口：向指定用户发送票号核销消息
*/
@PostMapping("/sendToUser")
public Result<String> sendToUser(@RequestBody Map<String, String> body) {
String userId = body.get("userId");
String ticketNo = body.get("ticketNo");
if (userId == null || ticketNo == null) return Result.fail("缺少 userId 或 ticketNo");
WebSocketMessage response = WebSocketMessage.info(ticketNo,
"这是一个测试消息，票号：" + ticketNo);
sessionService.sendToUser(userId, response);
// sessionService.sendVerificationSuccess(ticketNo,message);
return Result.success();
}
/** /**
* 测试接口：向指定用户发送票号核销消息 * 测试接口：向指定用户发送票号核销消息
*/ */
...@@ -39,9 +23,6 @@ public class WebSocketTestController { ...@@ -39,9 +23,6 @@ public class WebSocketTestController {
String message = body.get("message"); String message = body.get("message");
String ticketNo = body.get("ticketNo"); String ticketNo = body.get("ticketNo");
if (message == null || ticketNo == null) return Result.fail("缺少 message 或 ticketNo"); if (message == null || ticketNo == null) return Result.fail("缺少 message 或 ticketNo");
// WebSocketMessage response = WebSocketMessage.info(ticketNo,
// "这是一个测试消息，票号：" + ticketNo);
// sessionService.sendToUser(userId, response);
sessionService.sendVerificationSuccess(ticketNo,message); sessionService.sendVerificationSuccess(ticketNo,message);
return Result.success(); return Result.success();
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment