Commit beb3de1a authored by daniel's avatar daniel

fix(websocket): 两种方式的测试+联调

parent bb076059
...@@ -2,6 +2,7 @@ package com.dlbridge.jadp.config.web; ...@@ -2,6 +2,7 @@ package com.dlbridge.jadp.config.web;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry; import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
...@@ -12,8 +13,15 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { ...@@ -12,8 +13,15 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
@Override @Override
public void configureMessageBroker(MessageBrokerRegistry config) { public void configureMessageBroker(MessageBrokerRegistry config) {
// 启用简单的内存消息代理,指定消息代理目的地前缀 // 创建一个调度器,用于心跳定时任务
config.enableSimpleBroker("/topic", "/queue"); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
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.setApplicationDestinationPrefixes("/app");
// 设置用户目的地前缀(点对点消息) // 设置用户目的地前缀(点对点消息)
...@@ -24,7 +32,8 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { ...@@ -24,7 +32,8 @@ public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
public void registerStompEndpoints(StompEndpointRegistry registry) { public void registerStompEndpoints(StompEndpointRegistry registry) {
// 注册 STOMP 端点,指定使用 SockJS // 注册 STOMP 端点,指定使用 SockJS
registry.addEndpoint("/ws").setAllowedOrigins("*") // 允许所有来源,生产环境应指定具体域名 registry.addEndpoint("/ws").setAllowedOrigins("*") // 允许所有来源,生产环境应指定具体域名
.withSockJS(); .withSockJS()
.setHeartbeatTime(10000);
// 同时支持原生 WebSocket // 同时支持原生 WebSocket
registry.addEndpoint("/ws") registry.addEndpoint("/ws")
......
...@@ -28,6 +28,7 @@ public class BusController { ...@@ -28,6 +28,7 @@ public class BusController {
/** /**
* 查询个人购票列表 * 查询个人购票列表
*
* @return 个人购票列表 * @return 个人购票列表
*/ */
@ResponseBody @ResponseBody
...@@ -40,6 +41,7 @@ public class BusController { ...@@ -40,6 +41,7 @@ public class BusController {
/** /**
* 查询车票详情 * 查询车票详情
*
* @param param 查询参数 * @param param 查询参数
* @return 车票详情 * @return 车票详情
*/ */
...@@ -53,6 +55,7 @@ public class BusController { ...@@ -53,6 +55,7 @@ public class BusController {
/** /**
* 生成车票二维码 * 生成车票二维码
*
* @param cmptId 赛事ID * @param cmptId 赛事ID
* @param ticketNo 车票号 * @param ticketNo 车票号
* @return 乘车二维码图片 * @return 乘车二维码图片
...@@ -69,6 +72,7 @@ public class BusController { ...@@ -69,6 +72,7 @@ public class BusController {
/** /**
* 车票核销接口 * 车票核销接口
*
* @return 核销结果 * @return 核销结果
*/ */
@ResponseBody @ResponseBody
...@@ -80,6 +84,7 @@ public class BusController { ...@@ -80,6 +84,7 @@ public class BusController {
/** /**
* 查询赛事班车列表接口 * 查询赛事班车列表接口
*
* @return 班车列表 * @return 班车列表
*/ */
@ResponseBody @ResponseBody
...@@ -93,6 +98,7 @@ public class BusController { ...@@ -93,6 +98,7 @@ public class BusController {
/** /**
* 查询班次途经站点列表 * 查询班次途经站点列表
*
* @return 班次途经站点列表 * @return 班次途经站点列表
*/ */
@ResponseBody @ResponseBody
...@@ -105,6 +111,7 @@ public class BusController { ...@@ -105,6 +111,7 @@ public class BusController {
/** /**
* 查询赛事班车站点列表 * 查询赛事班车站点列表
*
* @return 班车站点列表 * @return 班车站点列表
*/ */
@ResponseBody @ResponseBody
...@@ -117,6 +124,7 @@ public class BusController { ...@@ -117,6 +124,7 @@ public class BusController {
/** /**
* 查询核销记录接口 * 查询核销记录接口
*
* @return 核销记录列表 * @return 核销记录列表
*/ */
@ResponseBody @ResponseBody
...@@ -129,6 +137,7 @@ public class BusController { ...@@ -129,6 +137,7 @@ public class BusController {
/** /**
* 赛事核销统计 * 赛事核销统计
*
* @param cmptId 赛事ID * @param cmptId 赛事ID
* @return 核销统计列表 * @return 核销统计列表
*/ */
...@@ -142,6 +151,7 @@ public class BusController { ...@@ -142,6 +151,7 @@ public class BusController {
/** /**
* 查询班车核销记录分页列表接口 * 查询班车核销记录分页列表接口
*
* @param queryDTO 查询参数 * @param queryDTO 查询参数
* @return 班车核销记录分页列表 * @return 班车核销记录分页列表
*/ */
......
...@@ -3,6 +3,7 @@ package com.dlbridge.jadp.websocket; ...@@ -3,6 +3,7 @@ package com.dlbridge.jadp.websocket;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.*; import java.util.*;
...@@ -31,13 +32,15 @@ public class WebSocketSessionService { ...@@ -31,13 +32,15 @@ public class WebSocketSessionService {
/** /**
* 注册用户会话 * 注册用户会话
*
* @param userId 用户ID(赛事ID+证件号,例:29643-HK002949527) * @param userId 用户ID(赛事ID+证件号,例:29643-HK002949527)
* @param ticketNo 票号 * @param ticketNo 票号
* @param sessionId 会话ID * @param sessionId 会话ID
*/ */
public void registerUserSession(String userId, String ticketNo, String sessionId) { public void registerUserSession(String userId, String ticketNo, String sessionId) {
// 更新用户-票号映射 // 更新用户-票号映射
userTicketsMap.computeIfAbsent(userId, k -> ConcurrentHashMap.newKeySet()).add(ticketNo); userTicketsMap.computeIfAbsent(userId, k -> ConcurrentHashMap.newKeySet())
.add(ticketNo);
// 更新票号-用户映射 // 更新票号-用户映射
ticketUserMap.put(ticketNo, userId); ticketUserMap.put(ticketNo, userId);
...@@ -46,15 +49,16 @@ public class WebSocketSessionService { ...@@ -46,15 +49,16 @@ public class WebSocketSessionService {
sessionUserMap.put(sessionId, userId); sessionUserMap.put(sessionId, userId);
// 更新会话-票号映射 // 更新会话-票号映射
sessionTicketsMap.computeIfAbsent(sessionId, k -> ConcurrentHashMap.newKeySet()).add(ticketNo); sessionTicketsMap.computeIfAbsent(sessionId, k -> ConcurrentHashMap.newKeySet())
.add(ticketNo);
log.info("用户会话注册成功 - 用户: {}, 票号: {}, 会话ID: {}", userId, ticketNo, sessionId); log.info("用户会话注册成功 - 用户: {}, 票号: {}, 会话ID: {}", userId, ticketNo, sessionId);
log.debug("当前会话状态 - 用户数: {}, 票号数: {}, 会话数: {}", log.debug("当前会话状态 - 用户数: {}, 票号数: {}, 会话数: {}", userTicketsMap.size(), ticketUserMap.size(), sessionUserMap.size());
userTicketsMap.size(), ticketUserMap.size(), sessionUserMap.size());
} }
/** /**
* 移除用户会话 * 移除用户会话
*
* @param sessionId 会话ID * @param sessionId 会话ID
*/ */
public void removeUserSession(String sessionId) { public void removeUserSession(String sessionId) {
...@@ -85,6 +89,7 @@ public class WebSocketSessionService { ...@@ -85,6 +89,7 @@ public class WebSocketSessionService {
/** /**
* 移除特定票号的会话 * 移除特定票号的会话
*
* @param sessionId 会话ID * @param sessionId 会话ID
* @param ticketNo 票号 * @param ticketNo 票号
*/ */
...@@ -156,7 +161,7 @@ public class WebSocketSessionService { ...@@ -156,7 +161,7 @@ public class WebSocketSessionService {
log.info("核销成功通知已发送 - 用户: {}, 票号: {}, 消息: {}", userId, ticketNo, message); log.info("核销成功通知已发送 - 用户: {}, 票号: {}, 消息: {}", userId, ticketNo, message);
// 核销成功后,移除该票号的会话映射 // 核销成功后,移除该票号的会话映射
removeTicketMappings(ticketNo); // removeTicketMappings(ticketNo);
} else { } else {
log.warn("未找到对应的用户会话,无法发送核销通知 - 票号: {}", ticketNo); log.warn("未找到对应的用户会话,无法发送核销通知 - 票号: {}", ticketNo);
} }
...@@ -201,7 +206,8 @@ public class WebSocketSessionService { ...@@ -201,7 +206,8 @@ public class WebSocketSessionService {
} }
// 从所有会话的票号集合中移除 // 从所有会话的票号集合中移除
sessionTicketsMap.values().forEach(tickets -> tickets.remove(ticketNo)); sessionTicketsMap.values()
.forEach(tickets -> tickets.remove(ticketNo));
log.info("票号映射关系已移除 - 票号: {}", ticketNo); log.info("票号映射关系已移除 - 票号: {}", ticketNo);
} }
...@@ -218,7 +224,8 @@ public class WebSocketSessionService { ...@@ -218,7 +224,8 @@ public class WebSocketSessionService {
stats.put("activeSessions", sessionTicketsMap.size()); stats.put("activeSessions", sessionTicketsMap.size());
// 计算平均每个用户的票号数 // 计算平均每个用户的票号数
double avgTicketsPerUser = userTicketsMap.values().stream() double avgTicketsPerUser = userTicketsMap.values()
.stream()
.mapToInt(Set::size) .mapToInt(Set::size)
.average() .average()
.orElse(0.0); .orElse(0.0);
......
package com.dlbridge.jadp.websocket; package com.dlbridge.jadp.websocket;
import com.dlbridge.jadp.common.web.Result;
import org.springframework.messaging.handler.annotation.MessageMapping; import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
...@@ -18,16 +19,31 @@ public class WebSocketTestController { ...@@ -18,16 +19,31 @@ public class WebSocketTestController {
/** /**
* 测试接口:向指定用户发送票号核销消息 * 测试接口:向指定用户发送票号核销消息
*/ */
@PostMapping("/sendTicket") @PostMapping("/sendToUser")
public String sendTicketMessage(@RequestBody Map<String, String> body) { public Result<String> sendToUser(@RequestBody Map<String, String> body) {
String userId = body.get("userId"); String userId = body.get("userId");
String ticketNo = body.get("ticketNo"); String ticketNo = body.get("ticketNo");
if (userId == null || ticketNo == null) return "缺少 userId 或 ticketNo"; if (userId == null || ticketNo == null) return Result.fail("缺少 userId 或 ticketNo");
WebSocketMessage response = WebSocketMessage.info(ticketNo, WebSocketMessage response = WebSocketMessage.info(ticketNo,
"这是一个测试消息,票号:" + ticketNo); "这是一个测试消息,票号:" + ticketNo);
sessionService.sendToUser(userId, response); sessionService.sendToUser(userId, response);
// sessionService.sendVerificationSuccess(ticketNo,message);
return Result.success();
}
/**
* 测试接口:向指定用户发送票号核销消息
*/
@PostMapping("/sendToVerify")
public Result<String> sendToVerify(@RequestBody Map<String, String> body) {
String message = body.get("message");
String ticketNo = body.get("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);
return "消息已发送到用户 " + userId; return Result.success();
} }
} }
\ No newline at end of file
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