userModelSessionMap = new ConcurrentHashMap<>();
+
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
sessions.put(session.getId(), session);
- log.info("WebSocket连接已建立，会话ID: {}", session.getId());
+
+ // 从会话属性中获取userId和modelId
+ Object userIdObj = session.getAttributes().get("userId");
+ Object modelIdObj = session.getAttributes().get("modelId");
+
+ Integer userId = null;
+ Integer modelId = null;
+
+ // 类型安全检查并转换
+ if (userIdObj instanceof Integer) {
+ userId = (Integer) userIdObj;
+ } else if (userIdObj != null) {
+ log.warn("userId不是预期的Integer类型，实际类型为: {}", userIdObj.getClass().getName());
+ }
+
+ if (modelIdObj instanceof Integer) {
+ modelId = (Integer) modelIdObj;
+ } else if (modelIdObj != null) {
+ log.warn("modelId不是预期的Integer类型，实际类型为: {}", modelIdObj.getClass().getName());
+ }
+
+ // 只有当userId和modelId都存在时才建立映射关系
+ if (userId != null && modelId != null) {
+ String key = userId + "_" + modelId;
+ userModelSessionMap.put(key, session.getId());
+ log.info("用户模型 {} 已注册到会话 {}", key, session.getId());
+ }
+
+ log.info("WebSocket连接已建立，会话ID: {}, 用户ID: {}, 模型ID: {}", session.getId(), userId, modelId);
// 发送连接成功的消息
JSONObject message = new JSONObject();
message.put("type", "connection");
message.put("status", "connected");
+ message.put("sessionId", session.getId());
session.sendMessage(new TextMessage(message.toJSONString()));
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
- sessions.remove(session.getId());
- log.info("WebSocket连接已关闭，会话ID: {}，状态: {}", session.getId(), status);
+ // 清理所有映射关系
+ String sessionId = session.getId();
+ sessions.remove(sessionId);
+
+ // 从用户模型映射中移除
+ userModelSessionMap.values().removeIf(id -> id.equals(sessionId));
+
+ log.info("WebSocket连接已关闭，会话ID: {}，状态: {}", sessionId, status);
}
- */
-/**
+ @Override
+ protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+ String payload = message.getPayload();
+ log.info("收到来自会话 {} 的消息: {}", session.getId(), payload);
+ }
+
+ /**
* 向所有连接的客户端发送消息
* @param message 消息内容
- *//*
-
+ */
public void sendToAll(String message) {
sessions.values().forEach(session -> {
try {
@@ -55,14 +99,12 @@ public class WebSocketService extends TextWebSocketHandler {
}
});
}
-
- */
-/**
+
+ /**
* 向指定会话发送消息
* @param sessionId 会话ID
* @param message 消息内容
- *//*
-
+ */
public void sendToSession(String sessionId, String message) {
WebSocketSession session = sessions.get(sessionId);
if (session != null && session.isOpen()) {
@@ -74,20 +116,36 @@ public class WebSocketService extends TextWebSocketHandler {
}
}
- */
-/**
- * 发送数据处理完成的通知
- * @param sessionId 会话ID
+ /**
+ * 向指定用户的模型发送消息
+ * @param userId 用户ID
+ * @param modelId 模型ID
+ * @param message 消息内容
+ */
+ public void sendToUserModel(Integer userId, Integer modelId, String message) {
+ String key = userId + "_" + modelId;
+ String sessionId = userModelSessionMap.get(key);
+ if (sessionId != null) {
+ sendToSession(sessionId, message);
+ } else {
+ log.warn("未找到用户 {} 的模型 {} 的WebSocket会话", userId, modelId);
+ }
+ }
+
+ /**
+ * 发送训练模型数据处理完成的通知
+ * @param userId 用户ID
+ * @param modelId 模型ID
* @param success 是否成功
* @param message 消息内容
- *//*
-
- public void sendDataProcessingNotification(String sessionId, boolean success, String message) {
+ */
+ public void sendDataProcessingNotification(Integer userId, Integer modelId, boolean success, String message) {
JSONObject notification = new JSONObject();
- notification.put("type", "dataProcessing");
+ notification.put("type", "modelDataProcessing");
+ notification.put("modelId", modelId);
notification.put("success", success);
notification.put("message", message);
- sendToSession(sessionId, notification.toJSONString());
+ sendToUserModel(userId, modelId, notification.toJSONString());
}
-}*/
+}
\ No newline at end of file
diff --git a/data/src/main/java/com/sdm/data/service/impl/ModelServiceImpl.java b/data/src/main/java/com/sdm/data/service/impl/ModelServiceImpl.java
index 7d114514..adff0050 100644
--- a/data/src/main/java/com/sdm/data/service/impl/ModelServiceImpl.java
+++ b/data/src/main/java/com/sdm/data/service/impl/ModelServiceImpl.java
@@ -120,6 +120,9 @@ public class ModelServiceImpl implements IModelService {
@Autowired
ITrainingModelAlgorithmParamService trainingModelAlgorithmParamService;
+ @Autowired
+ WebSocketService webSocketService;
+
@Override
@Transactional(rollbackFor = Exception.class)
@@ -204,8 +207,7 @@ public class ModelServiceImpl implements IModelService {
trainingModelService.lambdaUpdate().eq(TrainingModel::getId, trainingModelId).set(TrainingModel::getHandleStatus, "处理中").update();
// 异步调用Python脚本处理数据
- // String sessionId = handleLoadDataReq.getSessionId(); // 假设请求中包含sessionId
- processDataAsync(DATA_HANDLER_PYTHON_SCRIPT_PATH, paramJsonPath, trainingModelId, null);
+ processDataAsync(DATA_HANDLER_PYTHON_SCRIPT_PATH, paramJsonPath, trainingModelId, handleLoadDataReq.getUserId(),handleLoadDataReq.getTrainingModelId());
return SdmResponse.success("数据处理中");
} catch (Exception e) {
log.error("处理上传数据失败", e);
@@ -266,9 +268,10 @@ public class ModelServiceImpl implements IModelService {
*
* @param pythonScriptPath Python脚本路径
* @param paramJsonPath 参数文件路径
- * @param sessionId WebSocket会话ID
+ * @param userId 用户ID
+ * @param modelId 模型ID
*/
- private void processDataAsync(String pythonScriptPath, String paramJsonPath, Integer trainingModelId, String sessionId) {
+ private void processDataAsync(String pythonScriptPath, String paramJsonPath, Integer trainingModelId, Integer userId, Integer modelId) {
new Thread(() -> {
try {
// 调用Python脚本处理数据
@@ -347,16 +350,12 @@ public class ModelServiceImpl implements IModelService {
log.info("训练模型ID: {}数据处理完成,处理结果文件ID:{},总耗时: {} ms", trainingModelId, fileId, duration);
// 通过WebSocket通知前端处理完成
- /*if (sessionId != null) {
- webSocketService.sendDataProcessingNotification(sessionId, true, "数据处理完成");
- }*/
+ webSocketService.sendDataProcessingNotification(userId, modelId,true, "数据处理完成");
} catch (Exception e) {
trainingModelService.lambdaUpdate().eq(TrainingModel::getId, trainingModelId).set(TrainingModel::getHandleStatus, "失败").update();
log.error("异步处理数据失败", e);
// 通过WebSocket通知前端处理失败
- /*if (sessionId != null) {
- webSocketService.sendDataProcessingNotification(sessionId, false, "数据处理失败: " + e.getMessage());
- }*/
+ webSocketService.sendDataProcessingNotification(userId, modelId,false, "数据处理失败: " + e.getMessage());
}
}).start();
}
diff --git a/data/src/main/resources/static/websocketTest.html b/data/src/main/resources/static/websocketTest.html
new file mode 100644
index 00000000..7676b1e4
--- /dev/null
+++ b/data/src/main/resources/static/websocketTest.html
@@ -0,0 +1,94 @@
+
+
+
+
+ WebSocket 测试
+
+
+ WebSocket 测试页面
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/system/src/main/java/com/sdm/system/dao/SysUserRoleRelationMapper.java b/system/src/main/java/com/sdm/system/dao/SysUserRoleRelationMapper.java
new file mode 100644
index 00000000..e6817d37
--- /dev/null
+++ b/system/src/main/java/com/sdm/system/dao/SysUserRoleRelationMapper.java
@@ -0,0 +1,16 @@
+package com.sdm.system.dao;
+
+import com.sdm.system.model.entity.SysUserRoleRelation;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ *
+ * 用户角色关联表 Mapper 接口
+ *
+ *
+ * @author author
+ * @since 2025-10-23
+ */
+public interface SysUserRoleRelationMapper extends BaseMapper {
+
+}
diff --git a/system/src/main/java/com/sdm/system/model/entity/SysUserRoleRelation.java b/system/src/main/java/com/sdm/system/model/entity/SysUserRoleRelation.java
new file mode 100644
index 00000000..abb4c85c
--- /dev/null
+++ b/system/src/main/java/com/sdm/system/model/entity/SysUserRoleRelation.java
@@ -0,0 +1,49 @@
+package com.sdm.system.model.entity;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import java.time.LocalDateTime;
+import java.io.Serializable;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ *
+ * 用户角色关联表
+ *
+ *
+ * @author author
+ * @since 2025-10-23
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+@TableName("sys_user_role_relation")
+@ApiModel(value="SysUserRoleRelation对象", description="用户角色关联表")
+public class SysUserRoleRelation implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @ApiModelProperty(value = "关联记录主键ID")
+ @TableId(value = "id", type = IdType.AUTO)
+ private Integer id;
+
+ @ApiModelProperty(value = "用户ID（关联用户表主键）")
+ @TableField("user_id")
+ private Integer userId;
+
+ @ApiModelProperty(value = "角色ID（关联角色表主键）")
+ @TableField("role_id")
+ private Integer roleId;
+
+ @ApiModelProperty(value = "关联关系创建时间")
+ @TableField("add_time")
+ private LocalDateTime addTime;
+
+
+}
diff --git a/system/src/main/java/com/sdm/system/service/ISysUserRoleRelationService.java b/system/src/main/java/com/sdm/system/service/ISysUserRoleRelationService.java
new file mode 100644
index 00000000..60321281
--- /dev/null
+++ b/system/src/main/java/com/sdm/system/service/ISysUserRoleRelationService.java
@@ -0,0 +1,16 @@
+package com.sdm.system.service;
+
+import com.sdm.system.model.entity.SysUserRoleRelation;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ *
+ * 用户角色关联表 服务类
+ *
+ *
+ * @author author
+ * @since 2025-10-23
+ */
+public interface ISysUserRoleRelationService extends IService {
+
+}
diff --git a/system/src/main/java/com/sdm/system/service/impl/SysUserRoleRelationServiceImpl.java b/system/src/main/java/com/sdm/system/service/impl/SysUserRoleRelationServiceImpl.java
new file mode 100644
index 00000000..de36598c
--- /dev/null
+++ b/system/src/main/java/com/sdm/system/service/impl/SysUserRoleRelationServiceImpl.java
@@ -0,0 +1,20 @@
+package com.sdm.system.service.impl;
+
+import com.sdm.system.model.entity.SysUserRoleRelation;
+import com.sdm.system.dao.SysUserRoleRelationMapper;
+import com.sdm.system.service.ISysUserRoleRelationService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ *
+ * 用户角色关联表 服务实现类
+ *
+ *
+ * @author author
+ * @since 2025-10-23
+ */
+@Service
+public class SysUserRoleRelationServiceImpl extends ServiceImpl implements ISysUserRoleRelationService {
+
+}
diff --git a/system/src/main/resources/mapper/SysUserRoleRelationMapper.xml b/system/src/main/resources/mapper/SysUserRoleRelationMapper.xml
new file mode 100644
index 00000000..8ed68221
--- /dev/null
+++ b/system/src/main/resources/mapper/SysUserRoleRelationMapper.xml
@@ -0,0 +1,5 @@
+
+
+
+
+