mlx

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/consumer/ManualProjectConsumer.java

@@ -47,7 +47,11 @@ public class ManualProjectConsumer {
     @Autowired
     StringRedisTemplate redisTemplate;
     @Autowired
-    ProjectMapper projectMapper;
+    ManualProjectMapper manualProjectMapper;
+    @Autowired
+    AutoProjectMapper autoProjectMapper;
+    @Autowired
+    AutoSubProjectMapper autoSubProjectMapper;
     @Autowired
     TaskMapper taskMapper;
     @Autowired
@@ -64,7 +68,7 @@ public class ManualProjectConsumer {
     SensorOgtMapper sensorOgtMapper;
     @Autowired
     AlgorithmMapper algorithmMapper;
-//    @Autowired
+    //    @Autowired
 //    ApiClient apiClient;
     @Value("${scheduler.manual-project.topic}")
     String manualProjectTopic;
@@ -77,12 +81,6 @@ public class ManualProjectConsumer {
     @Value("${scheduler.linux-temp-path}")
     String linuxTempPath;
 
-
-    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "hello")
-    public void testConsumer(ConsumerRecord<String, String> projectRecord) {
-        log.info("------- testConsumer 消费成功：" + projectRecord.value());
-    }
-
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.topic}")
     @SneakyThrows
     public void parseManualProject(ConsumerRecord<String, String> projectRecord) {
@@ -101,7 +99,9 @@ public class ManualProjectConsumer {
         String projectJson = projectRecord.value();
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
         String projectId = projectMessageDTO.getProjectId();    // 项目 id
-        projectMapper.resetProjectState(projectId, DictConstants.PROJECT_RUNNING);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 方便测试。
+
+        redisTemplate.opsForValue().set(manualProjectTopic + ":" + projectId + ":taskCompleted", "0");  // 设置项目已完成任务为 0
+        manualProjectMapper.resetProjectState(projectId, DictConstants.PROJECT_RUNNING);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 方便测试。
         taskMapper.updateStateByProjectId(projectId, DictConstants.TASK_PENDING);    // 将该 project 下所有任务重置为待执行方便测试。
 
 
@@ -144,7 +144,7 @@ public class ManualProjectConsumer {
         int taskNumber = sceneList.size();
         Set<ScenePO> sceneSet = new HashSet<>(sceneList);
         log.info("------- ManualProjectConsumer 共有 " + taskNumber + " 个任务，对应 " + sceneSet.size() + " 个场景！");
-        projectMapper.updateTaskNumber(projectId, taskNumber);
+        manualProjectMapper.updateTaskNumber(projectId, taskNumber);
         // -------------------------------- 2 模型 --------------------------------
         // 根据车辆配置id vehicleConfigId, 获取 模型信息和传感器信息
         String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
@@ -440,7 +440,7 @@ public class ManualProjectConsumer {
 
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.stop-topic}")
     @SneakyThrows
-    public void stopProject(ConsumerRecord<String, String> stopRecord) {
+    public void stopManualProject(ConsumerRecord<String, String> stopRecord) {
         log.info("------- ManualProjectConsumer 接收到的项目终止消息为：" + stopRecord);
         //1 读取 kafka 的项目停止信息
         /*
@@ -452,8 +452,11 @@ public class ManualProjectConsumer {
         ObjectMapper objectMapper = new ObjectMapper();
         JsonNode jsonNode = objectMapper.readTree(json);
         String projectId = jsonNode.path("projectId").asText();
-        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_TERMINATED, TimeUtil.getNowForMysql());
+        manualProjectMapper.updateProjectState(projectId, DictConstants.PROJECT_TERMINATED, TimeUtil.getNowForMysql());
         LinuxUtil.execute("kubectl delete job project-" + projectId);
         redisTemplate.delete(manualProjectTopic + ":" + projectId + ":check");
     }
+
+
+
 }

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/AutoProjectMapper.java

@@ -0,0 +1,19 @@
+package com.css.simulation.resource.scheduler.mapper;
+
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
+
+@Mapper
+public interface AutoProjectMapper {
+
+
+    @Update("update simulation_automatic_project\n" +
+            "set now_run_state = #{nowRunState}\n" +
+            "where id = (select parent_id\n" +
+            "            from simulation_automatic_subproject\n" +
+            "            where id = #{subProjectId})")
+    void updateProjectStateBySubProjectId(@Param("subProjectId") String id, @Param("nowRunState") String state);
+
+}

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/AutoSubProjectMapper.java

@@ -0,0 +1,17 @@
+package com.css.simulation.resource.scheduler.mapper;
+
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
+
+@Mapper
+public interface AutoSubProjectMapper {
+
+    @Update("update simulation_automatic_subproject\n" +
+            "set now_run_state  = #{state}\n" +
+            "where id = #{id}")
+    void updateProjectStateById(@Param("subProjectId") String subProjectId, @Param("nowRunState") String nowRunState);
+
+
+}

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/ProjectMapper.java → simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/ManualProjectMapper.java

@@ -9,7 +9,7 @@ import java.sql.Timestamp;
 import java.util.List;
 
 @Mapper
-public interface ProjectMapper {
+public interface ManualProjectMapper {
 
 
     @Results(id = "project", value = {

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/scheduler/TickScheduler.java

@@ -2,7 +2,7 @@
 //
 //import api.common.pojo.constants.DictConstants;
 //import api.common.util.*;
-//import com.css.simulation.resource.scheduler.mapper.ProjectMapper;
+//import com.css.simulation.resource.scheduler.mapper.ManualProjectMapper;
 //import com.css.simulation.resource.scheduler.mapper.TaskMapper;
 //import com.css.simulation.resource.scheduler.pojo.po.TaskPO;
 //import com.css.simulation.resource.scheduler.service.TaskService;
@@ -35,7 +35,7 @@
 //    @Autowired
 //    TaskMapper taskMapper;
 //    @Autowired
-//    ProjectMapper projectMapper;
+//    ManualProjectMapper projectMapper;
 //    @Value("${scheduler.manual-project.job-yaml}")
 //    String jobYaml;
 //

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/service/TaskService.java

@@ -5,7 +5,7 @@ import api.common.util.*;
 import com.css.simulation.resource.scheduler.manager.TaskIndexManager;
 import com.css.simulation.resource.scheduler.manager.TaskManager;
 import com.css.simulation.resource.scheduler.mapper.IndexTemplateMapper;
-import com.css.simulation.resource.scheduler.mapper.ProjectMapper;
+import com.css.simulation.resource.scheduler.mapper.ManualProjectMapper;
 import com.css.simulation.resource.scheduler.mapper.TaskIndexMapper;
 import com.css.simulation.resource.scheduler.mapper.TaskMapper;
 import com.css.simulation.resource.scheduler.pojo.po.IndexTemplatePO;
@@ -52,7 +52,7 @@ public class TaskService {
     @Autowired
     StringRedisTemplate redisTemplate;
     @Autowired
-    ProjectMapper projectMapper;
+    ManualProjectMapper manualProjectMapper;
     //    @Autowired
 //    TaskManager taskManager;
     @Autowired
@@ -100,7 +100,7 @@ public class TaskService {
     public void taskTick(String taskId) {
         log.info("TaskService--taskTick 任务 " + taskId + "心跳！");
         // 刷新 redis 心跳时间
-        ProjectPO projectPO = projectMapper.selectById(taskId);
+        ProjectPO projectPO = manualProjectMapper.selectById(taskId);
         String projectId = projectPO.getId();
         redisTemplate.opsForValue().set(manualProjectTopic + ":" + projectId + ":" + taskId, TimeUtil.getNowString());
     }
@@ -164,7 +164,7 @@ public class TaskService {
             taskMapper.updateFailStateWithStopTime(taskId, state, TimeUtil.getNowForMysql(), DictConstants.TASK_ERROR_REASON_4);
             redisTemplate.delete("podName:" + taskId);
         }
-        ProjectPO projectPO = projectMapper.selectById(taskId);
+        ProjectPO projectPO = manualProjectMapper.selectById(taskId);
         if (projectPO == null) {
             session.close();
             client.stop();
@@ -175,9 +175,9 @@ public class TaskService {
         redisTemplate.delete(keys);
         String scenePackageId = projectPO.getScenePackageId();  // 场景测试包 id，指标根 id
 //        log.info("------- /state 任务 " + taskId + " 的父项目为：" + projectId);
-        int taskNum = projectMapper.selectTaskNumById(projectId);
-        int endTaskNum = projectMapper.selectEndTaskNum(projectId);    // 查询已结束的任务 'Aborted', 'PendingAnalysis', 'Terminated'
-        projectMapper.updateTaskCompleted(projectId, endTaskNum);
+        int taskNum = manualProjectMapper.selectTaskNumById(projectId);
+        int endTaskNum = manualProjectMapper.selectEndTaskNum(projectId);    // 查询已结束的任务 'Aborted', 'PendingAnalysis', 'Terminated'
+        manualProjectMapper.updateTaskCompleted(projectId, endTaskNum);
         log.info("TaskService--state 项目 " + projectId + " 完成进度为：" + endTaskNum + "/" + taskNum);
         if (taskNum != endTaskNum) {  // 已结束任务数等于所有任务数量，才会准备打分；否则退出。
             session.close();
@@ -348,7 +348,7 @@ public class TaskService {
         String post = HttpUtil.post(closeableHttpClient, requestConfig, evaluationLevelUri, headers, params);
         log.info("TaskService--state 访问仿真云平台评价等级接口：" + evaluationLevelUri + "，请求头为：" + headers + "，请求体为：" + params + "结果为：" + post);
         log.info("TaskService--state 项目 " + projectId + " 打分完成！");
-        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED, TimeUtil.getNowForMysql());   // 修改该 project 的状态为已完成
+        manualProjectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED, TimeUtil.getNowForMysql());   // 修改该 project 的状态为已完成
         log.info("TaskService--state 项目 " + projectId + " 执行完成！");
         session.close();
         client.stop();