opensource/RuoYi-Vue-Plus
2
0
Fork 1
mirror of https://gitee.com/dromara/RuoYi-Vue-Plus.git synced 2026-09-19 01:48:47 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

设备数据提送并发问题优化

Browse Source
This commit is contained in:
zzc 2025-01-12 00:59:40 +08:00
parent fa8208d6a5
commit 22fd28c891
3 changed files with 56 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ruoyi-admin/src/main/java/org/dromara/web/utils/WebSocketBusinessHandler.java
View File

@ -78,7 +78,8 @@ public class WebSocketBusinessHandler implements WebSocketApi {
// EqEquipmentDetailVo data = deviceMsg.getData(); // EqEquipmentDetailVo data = deviceMsg.getData();
sendMsg = JsonUtils.toJsonString(deviceMsg); sendMsg = JsonUtils.toJsonString(deviceMsg);
log.info("socket详情发送：{}",sendMsg); log.info("socket详情发送：{}",sendMsg);
sendMsg(sendMsg, userId); // sendMsg(sendMsg, userId);
sendMsg(sendMsg, Long.valueOf("5"+userId.toString().substring(1)));
} }
private void initLanguage(Long userId) { private void initLanguage(Long userId) {
@ -131,7 +132,8 @@ public class WebSocketBusinessHandler implements WebSocketApi {
} }
sendMsg = isTure? JsonUtils.toJsonString(list):sendMsg; sendMsg = isTure? JsonUtils.toJsonString(list):sendMsg;
log.info("socket列表发送：{}",sendMsg); log.info("socket列表发送：{}",sendMsg);
sendMsg(sendMsg, userId); // sendMsg(sendMsg, userId);
sendMsg(sendMsg, Long.valueOf("6"+userId.toString().substring(1)));
} }

48
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java
View File

@ -37,13 +37,23 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
log.info("[connect] invalid token received. sessionId: {}", session.getId()); log.info("[connect] invalid token received. sessionId: {}", session.getId());
return; return;
} }
WebSocketSessionHolder.addSession(loginUser.getUserId(), session); String currentPath = session.getUri().getPath();
log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType()); Long userId = loginUser.getUserId();
if (currentPath.contains(WebSocketConstants.detail)){
WebSocketSessionHolder.addSession(Long.valueOf("5"+userId.toString().substring(1) ), session);
}else if (currentPath.contains(WebSocketConstants.list)){
WebSocketSessionHolder.addSession(Long.valueOf("6"+userId.toString().substring(1)), session);
}else{
WebSocketSessionHolder.addSession(userId, session);
}
log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), userId, loginUser.getUserType());
} }
@Autowired @Autowired
WebSocketApi webSocketApi; WebSocketApi webSocketApi;
/** /**
* 处理接收到的文本消息 * 处理接收到的文本消息
* *
@ -57,10 +67,10 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
log.info("socket：接受到消息 {}", message.getPayload()); log.info("socket：接受到消息 {}", message.getPayload());
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY); LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
String currentPath = session.getUri().getPath(); String currentPath = session.getUri().getPath();
String lastPath = WebSocketSessionHolder.getSessions(loginUser.getUserId()).getUri().getPath(); /*String lastPath = WebSocketSessionHolder.getSessions(loginUser.getUserId()).getUri().getPath();
if (!currentPath.equals(lastPath)){ if (!currentPath.equals(lastPath)){
WebSocketSessionHolder.addSession(loginUser.getUserId(), session); WebSocketSessionHolder.addSession(loginUser.getUserId(), session);
} }*/
String payload = message.getPayload(); String payload = message.getPayload();
if (currentPath.contains(WebSocketConstants.detail)){ if (currentPath.contains(WebSocketConstants.detail)){
webSocketApi.dealWithDetailMsg(loginUser, payload); webSocketApi.dealWithDetailMsg(loginUser, payload);
@ -111,7 +121,11 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
*/ */
@Override @Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
// 当WebSocket发生传输错误时调用，比如网络问题导致连接断开
// 这里可以认为是客户端断开连接，因为服务端无法主动抛出传输错误
log.info("websocket连接-网络-客户端断开连接，错误信息：" + exception.getMessage());
log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage()); log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage());
super.handleTransportError(session, exception);
} }
/** /**
@ -121,14 +135,33 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
* @param status 关闭状态信息 * @param status 关闭状态信息
*/ */
@Override @Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY); LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
if (ObjectUtil.isNull(loginUser)) { if (ObjectUtil.isNull(loginUser)) {
log.info("[disconnect] invalid token received. sessionId: {}", session.getId()); log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
return; return;
} }
WebSocketSessionHolder.removeSession(loginUser.getUserId()); // 当WebSocket连接关闭时调用
log.info("[disconnect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType()); if (status.equals(CloseStatus.SERVER_ERROR)) {
// 服务端断开连接，比如服务端代码中调用了session.close()
log.info("websocket连接-服务端-断开连接，关闭状态：" + status);
} else {
// 客户端断开连接，比如客户端调用了close方法
log.info("websocket连接-客户端-断开连接，关闭状态：" + status);
}
super.afterConnectionClosed(session, status);
String currentPath = session.getUri().getPath();
Long userId = loginUser.getUserId();
if (currentPath.contains(WebSocketConstants.detail)){
WebSocketSessionHolder.removeSession(Long.valueOf("5"+userId.toString().substring(1)));
}else if (currentPath.contains(WebSocketConstants.list)){
WebSocketSessionHolder.removeSession(Long.valueOf("6"+userId.toString().substring(1)));
}else{
WebSocketSessionHolder.removeSession(userId);
}
// WebSocketSessionHolder.removeSession(loginUser.getUserId());
log.info("[disconnect] sessionId: {},userId:{},userType:{}", session.getId(), userId, loginUser.getUserType());
} }
/** /**
@ -141,4 +174,5 @@ public class PlusWebSocketHandler extends AbstractWebSocketHandler {
return false; return false;
} }
} }

19
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java
View File

@ -106,7 +106,7 @@ public class WebSocketUtils {
public static void sendMessage(WebSocketSession session, String message) { public static void sendMessage(WebSocketSession session, String message) {
sendMessage(session, new TextMessage(message)); sendMessage(session, new TextMessage(message));
} }
private static final Object sessionLock = new Object();
/** /**
* 向指定的WebSocket会话发送WebSocket消息对象 * 向指定的WebSocket会话发送WebSocket消息对象
* *
@ -114,14 +114,17 @@ public class WebSocketUtils {
* @param message 要发送的WebSocket消息对象 * @param message 要发送的WebSocket消息对象
*/ */
private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) { private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
if (session == null || !session.isOpen()) { synchronized (sessionLock) {
log.warn("[send] session会话已经关闭"); if (session == null || !session.isOpen()) {
} else { log.warn("[send] session会话已经关闭");
try { } else {
session.sendMessage(message); try {
} catch (IOException e) { session.sendMessage(message);
log.error("[send] session({}) 发送消息({}) 异常", session, message, e); } catch (IOException e) {
log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
}
} }
} }
} }
} }