mlx

api-common/src/main/java/api/common/pojo/constants/DictConstants.java

@@ -98,7 +98,7 @@ public class DictConstants {
     public static final String ROLE_CODE_SUBUESR = "3"; //用户类型-子账户
 
 
-    //任务失败原因
+    // 任务失败原因
     public static final String TASK_ERROR_REASON_1 = "任务执行超时！";
     public static final String TASK_ERROR_REASON_2 = "任务执行超时或未知错误！";
     public static final String TASK_ERROR_REASON_3 = "容器终止！";
@@ -108,11 +108,18 @@ public class DictConstants {
     public static final String USE_TYPE_EXCLUSIVE = "1";//独占使用
     public static final String USE_TYPE_PUBLIC = "2";//公共使用
 
-    //场景上传任务状态
+    // 场景上传任务状态
     public static final String SCENE_IMPORT_STATUS_0 = "0"; //未上传
     public static final String SCENE_IMPORT_STATUS_1 = "1"; //上传中
     public static final String SCENE_IMPORT_STATUS_2 = "2"; //上传完成
     public static final String SCENE_IMPORT_STATUS_3 = "3"; //解析中
     public static final String SCENE_IMPORT_STATUS_4 = "4"; //解析完成
 
+    // 项目类型
+    public static final String PROJECT_TYPE_MANUAL = "1"; // 手动运行项目
+    public static final String PROJECT_TYPE_AUTO_SUB = "2"; // 自动运行项目子项目
+
+    // 集群id
+    public static final String SYSTEM_CLUSTER_ID = "system"; // 超管使用此集群id执行项目
+
 }

api-common/src/main/java/api/common/pojo/dto/ProjectMessageDTO.java

@@ -25,7 +25,8 @@ public class ProjectMessageDTO {
     private String algorithmId;// 算法 id
     private String vehicleConfigId;// 车辆配置 id
     private String scenePackageId;// 场景包 id
-    private int maxSimulationTime;// 最大仿真时间（秒）
-    private int parallelism;// 并行度，创建 pod 时使用
+    private Long maxSimulationTime;// 最大仿真时间（秒）
+    private Long parallelism;// 并行度，创建 pod 时使用
+    private String projectType;// 项目类型
 
 }

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

@@ -1,219 +0,0 @@
-//package com.css.simulation.resource.scheduler.consumer;
-//
-//
-//import api.common.pojo.constants.DictConstants;
-//import api.common.pojo.dto.ProjectMessageDTO;
-//import api.common.util.*;
-//import com.css.simulation.resource.scheduler.mapper.*;
-//import com.css.simulation.resource.scheduler.pojo.po.*;
-//import com.css.simulation.resource.scheduler.service.ManualProjectService;
-//import com.fasterxml.jackson.databind.JsonNode;
-//import com.fasterxml.jackson.databind.ObjectMapper;
-//import lombok.SneakyThrows;
-//import lombok.extern.slf4j.Slf4j;
-//import org.apache.kafka.clients.consumer.ConsumerRecord;
-//import org.apache.sshd.client.SshClient;
-//import org.apache.sshd.client.session.ClientSession;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.beans.factory.annotation.Value;
-//import org.springframework.data.redis.core.StringRedisTemplate;
-//import org.springframework.kafka.annotation.KafkaListener;
-//import org.springframework.kafka.core.KafkaTemplate;
-//import org.springframework.stereotype.Component;
-//
-//import java.util.HashSet;
-//import java.util.List;
-//import java.util.Set;
-//
-//@Component
-//@Slf4j
-//public class AutoProjectConsumer {
-//
-//    @Autowired
-//    KafkaTemplate<String, String> kafkaTemplate;
-//    @Autowired
-//    StringRedisTemplate redisTemplate;
-//    @Autowired
-//    AutoProjectMapper autoProjectMapper;
-//    @Autowired
-//    AutoSubProjectMapper autoSubProjectMapper;
-//    @Autowired
-//    TaskMapper taskMapper;
-//    @Autowired
-//    IndexMapper indexMapper;
-//    @Autowired
-//    IndexTemplateMapper indexTemplateMapper;
-//    @Autowired
-//    SceneMapper sceneMapper;
-//    @Autowired
-//    VehicleMapper vehicleMapper;
-//    @Autowired
-//    SensorCameraMapper sensorCameraMapper;
-//    @Autowired
-//    SensorOgtMapper sensorOgtMapper;
-//    @Autowired
-//    AlgorithmMapper algorithmMapper;
-//    @Autowired
-//    UserMapper userMapper;
-//    @Autowired
-//    ClusterMapper clusterMapper;
-//    @Autowired
-//    ManualProjectService manualProjectService;
-//    //    @Autowired
-////    ApiClient apiClient;
-//    @Value("${scheduler.manual-project.topic}")
-//    String manualProjectTopic;
-//    @Value("${scheduler.manual-project.result-path-minio}")
-//    String resultPathMinio;
-//    @Value("${scheduler.manual-project.job-template}")
-//    String jobTemplate;
-//    @Value("${scheduler.manual-project.job-yaml}")
-//    String jobYaml;
-//
-//    @Value("${scheduler.score.hostname}")
-//    String hostname;
-//    @Value("${scheduler.score.username}")
-//    String username;
-//    @Value("${scheduler.score.password}")
-//    String password;
-//
-//
-//    /**
-//     * 任务运行前首先判断用户是否拥有可分配资源
-//     *
-//     * @param projectRecord 项目启动消息
-//     */
-//    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.auto-project.topic}")
-//    @SneakyThrows
-//    public void cacheAutoProject(ConsumerRecord<String, String> projectRecord) {
-//        log.info("------- AutoProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectRecord);
-//        String projectJson = projectRecord.value();
-//        //1 读取 kafka 的 project 信息
-//        ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
-//        String subProjectId = projectMessageDTO.getProjectId();    // 项目 id
-//        String packageId = projectMessageDTO.getScenePackageId();   // 场景测试包 id
-//        int maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
-//        String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
-//        String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
-//        //2 根据 projectId 获取创建用户 id
-//        String userId = autoSubProjectMapper.selectCreateUserById(subProjectId);
-//        //3 获取用户类型（管理员账户、管理员子账户、普通账户、普通子账户）（独占、共享）
-//        UserPO userPO = userMapper.selectById(userId);
-//        String roleCode = userPO.getRoleCode();
-//        String useType = userPO.getUseType();
-//        ClusterPO clusterPO;
-//        if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
-//            parseAutoProject(projectJson);
-//            return;
-//        } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
-//            // 获取拥有的节点数量，即仿真软件证书数量
-//            clusterPO = clusterMapper.selectByUserId(userId);
-//        } else if (DictConstants.ROLE_CODE_SUBUESR.equals(roleCode)) {
-//            if (DictConstants.USE_TYPE_EXCLUSIVE.equals(useType)) {   //3-2 普通子账户，根据自己的独占节点排队
-//                clusterPO = clusterMapper.selectByUserId(userId);
-//            } else {    // 共享子账户需要查询父账户的集群 id
-//                String parentUserId = userPO.getCreateUserId();
-//                clusterPO = clusterMapper.selectByUserId(parentUserId);
-//            }
-//        } else {
-//            parseAutoProject(projectJson);
-//            return;
-//        }
-//        // 获取拥有的节点数量，即仿真软件证书数量
-//        String clusterId = clusterPO.getId();
-//        int simulationLicenseNumber = clusterPO.getNumSimulationLicense();
-//        // 获取该集群中正在运行的项目数量
-//        Set<String> runningProjectSet = redisTemplate.keys(manualProjectTopic + ":cluster:" + clusterId + ":running" + "*");
-//        int runningProjectNumber = CollectionUtil.isEmpty(runningProjectSet) ? 0 : runningProjectSet.size();
-//        if (runningProjectNumber < simulationLicenseNumber) {
-//            redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running" + projectId, projectJson);
-//            parseAutoProject(projectJson);
-//        } else {
-//            redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":waiting" + projectId, projectJson);
-//        }
-//    }
-//
-//
-//    /**
-//     * 开始执行以及重新执行
-//     *
-//     * @param projectJson 项目启动消息
-//     */
-//    @SneakyThrows
-//    public void parseAutoProject( String projectJson) {
-//
-//        // -------------------------------- 0 准备 --------------------------------
-//        log.info("------- ManualProjectConsumer 接收到项目开始消息为：" + projectJson);
-//        //1 读取 kafka 的 project 信息
-//        ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
-//        String projectId = projectMessageDTO.getProjectId();    // 项目 id
-//        String packageId = projectMessageDTO.getScenePackageId();   // 场景测试包 id
-//        int maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
-//        String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
-//        String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
-//        String userId = manualProjectMapper.selectCreateUserById(projectId);
-//        //2 执行前准备，删除改项目下所有任务，即重新执行改项目时需要新的测试包
-//        manualProjectService.prepare(manualProjectTopic, userId, projectId, projectJson);
-//        // -------------------------------- 1 查询场景 --------------------------------
-//        //1-1 根据场景测试包 packageId，拿到场景集合（不包括重复场景），重复场景会在发送消息时根据叶子指标发送多次。
-//        List<ScenePO> scenePOList = manualProjectService.handlePackage(manualProjectTopic, userId, projectId, packageId);
-//        Set<ScenePO> scenePOSet = new HashSet<>(scenePOList);
-//        // -------------------------------- 2 查询模型 --------------------------------
-//        //2-1 根据车辆配置id vehicleConfigId, 获取 模型信息和传感器信息
-//        VehiclePO vehiclePO = vehicleMapper.selectByVehicleConfigId(vehicleConfigId);   // 车辆
-//        List<CameraPO> cameraPOList = sensorCameraMapper.selectCameraByVehicleConfigId(vehicleConfigId);    // 摄像头
-//        List<OgtPO> ogtPOList = sensorOgtMapper.selectOgtByVehicleId(vehicleConfigId); // 完美传感器
-//        // -------------------------------- 3 发送任务消息 --------------------------------
-//        manualProjectService.sendTaskMessage(manualProjectTopic, userId, projectId, maxSimulationTime, scenePOSet, vehiclePO, cameraPOList, ogtPOList);
-//        // -------------------------------- 4 算法导入（一期按单机版做） --------------------------------
-//        String algorithmDockerImage = manualProjectService.handleAlgorithm(projectId, algorithmId);
-//        // -------------------------------- 5 创建 pod 开始执行 --------------------------------
-//        int completions = scenePOList.size();     // 结束标
-//        int parallelism = projectMessageDTO.getParallelism();    // 并行度
-//        log.info("------- ManualProjectConsumer 项目 " + projectId + " 的完成度为：" + completions);
-//        log.info("------- ManualProjectConsumer 项目 " + projectId + " 的并行度为：" + parallelism);
-//        String jobTemplateYamlPathSource = jobTemplate + "job-template.yaml";
-//        String jobTemplateYamlPathTarget = jobYaml + "project-" + projectId + ".yaml";
-//        String yamlSource = FileUtil.read(jobTemplateYamlPathSource);
-//        log.info("------- ManualProjectConsumer 模板文件为：" + yamlSource);
-//        String replace0 = yamlSource.replace("job-cloud-simulation", "project-" + projectId);
-//        String replace1 = replace0.replace("vtd-container", "vtd-" + projectId);
-//        String replace2 = replace1.replace("algorithm-container", "algorithm-" + projectId);
-//        String replace3 = replace2.replace("algorithm-image", algorithmDockerImage);
-//        String replace4 = replace3.replace("projectId", projectId);
-//        String replace5 = replace4.replace("completions-number", completions + "");
-//        String replace6 = replace5.replace("parallelism-number", parallelism + "");
-//        String replace7 = replace6.replace("apiVers1on", "apiVersion");
-//        String replace8 = replace7.replace("1atch/v1", "batch/v1");
-//        log.info("------- ManualProjectConsumer 开始执行 yaml 文件" + replace8);
-//        FileUtil.writeStringToLocalFile(replace8, jobTemplateYamlPathTarget);
-//
-//
-//        SshClient client = SshUtil.getClient();
-//        ClientSession session = SshUtil.getSession(client, hostname, username, password);
-//        SshUtil.execute(session, "kubectl apply -f " + jobTemplateYamlPathTarget);
-//        session.close();
-//        client.stop();
-//    }
-//
-//    @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.stop-topic}")
-//    @SneakyThrows
-//    public void stopManualProject(ConsumerRecord<String, String> stopRecord) {
-//        log.info("ManualProjectConsumer--stopManualProject 接收到的项目终止消息为：" + stopRecord);
-//        //1 读取 kafka 的项目停止信息
-//        /*
-//            {
-//                "projectId": "sadfasdfs",	// 项目 id
-//            }
-//         */
-//        String json = stopRecord.value();
-//        ObjectMapper objectMapper = new ObjectMapper();
-//        JsonNode jsonNode = objectMapper.readTree(json);
-//        String projectId = jsonNode.path("projectId").asText();
-//        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/consumer/ManualProjectConsumer.java → simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/consumer/ProjectConsumer.java

@@ -27,7 +27,7 @@ import java.util.Set;
 
 @Component
 @Slf4j
-public class ManualProjectConsumer {
+public class ProjectConsumer {
 
     @Autowired
     KafkaTemplate<String, String> kafkaTemplate;
@@ -88,68 +88,96 @@ public class ManualProjectConsumer {
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.topic}")
     @SneakyThrows
     public void cacheManualProject(ConsumerRecord<String, String> projectRecord) {
-        log.info("ManualProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectRecord);
         String projectJson = projectRecord.value();
+        log.info("ProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectJson);
         //1 读取 kafka 的 project 信息
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
-        String projectId = projectMessageDTO.getProjectId();    // 项目 id
-        String packageId = projectMessageDTO.getScenePackageId();   // 场景测试包 id
-        int maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
-        String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
-        String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
-        int parallelism = projectMessageDTO.getParallelism();
+        String projectId = projectMessageDTO.getProjectId();    // 手动执行项目 id 或 自动执行子项目 id
+        Long parallelism = projectMessageDTO.getParallelism();   // 项目并行度
+        String projectType = projectMessageDTO.getProjectType(); // 项目类型
         //2 根据 projectId 获取创建用户 id
-        String userId = manualProjectMapper.selectCreateUserById(projectId);
+        String userId;
+        if (DictConstants.PROJECT_TYPE_MANUAL.equals(projectType)) {
+            userId = manualProjectMapper.selectCreateUserById(projectId);
+        } else if (DictConstants.PROJECT_TYPE_AUTO_SUB.equals(projectType)) {
+            userId = autoSubProjectMapper.selectCreateUserById(projectId);
+        } else {
+            log.error("ProjectConsumer--cacheManualProject 项目类型错误：" + projectJson);
+            return;
+        }
+        if (StringUtil.isEmpty(userId)) {
+            log.error("ProjectConsumer--cacheManualProject 未查询到项目创建人：" + projectJson);
+            return;
+        }
         //3 获取用户类型（管理员账户、管理员子账户、普通账户、普通子账户）（独占、共享）
         UserPO userPO = userMapper.selectById(userId);
         String roleCode = userPO.getRoleCode();
         String useType = userPO.getUseType();
         ClusterPO clusterPO;
+        String clusterPrefix;
+        String clusterRunningPrefix;
+        String clusterWaitingPrefix;
+        String projectRunningKey;
+        String projectWaitingKey;
         if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
-            parseManualProject(projectJson);
+            clusterPrefix = "cluster:" + DictConstants.SYSTEM_CLUSTER_ID;
+            clusterRunningPrefix = clusterPrefix + ":running";
+            projectRunningKey = clusterRunningPrefix + ":" + projectId;
+            run(DictConstants.SYSTEM_CLUSTER_ID, projectId, projectRunningKey, projectJson);
             return;
         } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
-            // 获取拥有的节点数量，即仿真软件证书数量
             clusterPO = clusterMapper.selectByUserId(userId);
         } else if (DictConstants.ROLE_CODE_SUBUESR.equals(roleCode)) {
-            if (DictConstants.USE_TYPE_EXCLUSIVE.equals(useType)) {   //3-2 普通子账户，根据自己的独占节点排队
+            if (DictConstants.USE_TYPE_EXCLUSIVE.equals(useType)) {   //3-3 普通子账户，根据自己的独占节点排队
                 clusterPO = clusterMapper.selectByUserId(userId);
-            } else {    // 共享子账户需要查询父账户的集群 id
+            } else {    //3-4 共享子账户，根据父账户的共享节点排队
                 String parentUserId = userPO.getCreateUserId();
                 clusterPO = clusterMapper.selectByUserId(parentUserId);
             }
         } else {
-            parseManualProject(projectJson);
+            log.error("ProjectConsumer--cacheManualProject 未知账户类型，不予执行：" + projectJson);
             return;
         }
         // 获取拥有的节点数量，即仿真软件证书数量
         String clusterId = clusterPO.getId();
         int simulationLicenseNumber = clusterPO.getNumSimulationLicense();
-        // 获取该集群中正在运行的项目
-        Set<String> runningProjectSet = redisTemplate.keys(manualProjectTopic + ":cluster:" + clusterId + ":running" + "*");
-        // 根据项目 json 获取每个项目的并行度
-        if (CollectionUtil.isNotEmpty(runningProjectSet)) {
-            long parallelismSum = 0;
-            for (String projectKey : runningProjectSet) {
-                String projectJsonTemp = redisTemplate.opsForValue().get(projectKey);
-                ProjectMessageDTO projectMessageTemp = JsonUtil.jsonToBean(projectJsonTemp, ProjectMessageDTO.class);
-                parallelismSum += projectMessageTemp.getParallelism();
-            }
-            if (parallelismSum + parallelism <= simulationLicenseNumber) {
-                log.info("ManualProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 执行！");
-                redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId, projectJson);
-                parseManualProject(projectJson);
-            } else {
-                log.info("ManualProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 放入等待队列！");
-                redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + projectId, projectJson);
-            }
+        // 获取该集群中正在运行的项目，如果没有则立即执行
+        clusterPrefix = "cluster:" + clusterId;
+        clusterRunningPrefix = clusterPrefix + ":running";
+        clusterWaitingPrefix = clusterPrefix + ":waiting";
+        projectRunningKey = clusterRunningPrefix + ":" + projectId;
+        projectWaitingKey = clusterWaitingPrefix + ":" + projectId;
+        Set<String> runningProjectSet = redisTemplate.keys(clusterRunningPrefix + "*");
+        if (CollectionUtil.isEmpty(runningProjectSet)) {
+            run(clusterId, projectId, projectRunningKey, projectJson);
+            return;
+        }
+        // 计算正在运行的项目的并行度总和
+        long parallelismSum = 0;
+        for (String projectKey : runningProjectSet) {
+            String projectJsonTemp = redisTemplate.opsForValue().get(projectKey);
+            ProjectMessageDTO projectMessageTemp = JsonUtil.jsonToBean(projectJsonTemp, ProjectMessageDTO.class);
+            parallelismSum += projectMessageTemp.getParallelism();
+        }
+        // 如果执行后的并行度总和小于最大节点数则执行，否则不执行
+        if (parallelismSum + parallelism <= simulationLicenseNumber) {
+            run(clusterId, projectId, projectRunningKey, projectJson);
         } else {
-            log.info("ManualProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 执行！");
-            redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId, projectJson);
-            parseManualProject(projectJson);
+            wait(clusterId, projectId, projectWaitingKey, projectJson);
         }
     }
 
+    public void run(String clusterId, String projectId, String projectRunningKey, String projectJson) {
+        log.info("ProjectConsumer--run 集群 " + clusterId + " 将项目 " + projectId + " 执行！");
+        redisTemplate.opsForValue().set(projectRunningKey, projectJson);
+        parseManualProject(projectJson, "cluster:" + clusterId, projectRunningKey);
+    }
+
+    public void wait(String clusterId, String projectId, String projectWaitingKey, String projectJson) {
+        log.info("ProjectConsumer--cacheManualProject 集群 " + clusterId + " 将项目 " + projectId + " 放入等待队列！");
+        redisTemplate.opsForValue().set(projectWaitingKey, projectJson);
+    }
+
 
     /**
      * 开始执行以及重新执行
@@ -157,42 +185,42 @@ public class ManualProjectConsumer {
      * @param projectJson 项目启动消息
      */
     @SneakyThrows
-    public void parseManualProject(String projectJson) {
+    public void parseManualProject(String projectJson, String clusterPrefix, String projectRunningPrefix) {
 
         // -------------------------------- 0 准备 --------------------------------
-        log.info("------- ManualProjectConsumer 接收到项目开始消息为：" + projectJson);
+        log.info("ProjectConsumer--parseManualProject 接收到项目开始消息为：" + projectJson);
         //1 读取 kafka 的 project 信息
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
         String projectId = projectMessageDTO.getProjectId();    // 项目 id
         String packageId = projectMessageDTO.getScenePackageId();   // 场景测试包 id
-        int maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
+        Long maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
         String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
         String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
         String userId = manualProjectMapper.selectCreateUserById(projectId);
-        int parallelism = projectMessageDTO.getParallelism();    // 并行度
+        Long parallelism = projectMessageDTO.getParallelism();    // 并行度
         //2 执行前准备，删除改项目下所有任务，即重新执行改项目时需要新的测试包
-        manualProjectService.prepare(manualProjectTopic, userId, projectId, projectJson);
+        manualProjectService.prepare(clusterPrefix, projectId, projectJson);
         // -------------------------------- 1 查询场景 --------------------------------
         //1-1 根据场景测试包 packageId，拿到场景集合（不包括重复场景），重复场景会在发送消息时根据叶子指标发送多次。
-        List<ScenePO> scenePOList = manualProjectService.handlePackage(manualProjectTopic, userId, projectId, packageId);
-        Set<ScenePO> scenePOSet = new HashSet<>(scenePOList);
+        List<ScenePO> scenePOList = manualProjectService.handlePackage(projectRunningPrefix, projectId, packageId);
+        Set<ScenePO> scenePOSet = new HashSet<>(scenePOList); // 如果不去重的话会出现多个场景重复关联多个指标
         // -------------------------------- 2 查询模型 --------------------------------
         //2-1 根据车辆配置id vehicleConfigId, 获取 模型信息和传感器信息
         VehiclePO vehiclePO = vehicleMapper.selectByVehicleConfigId(vehicleConfigId);   // 车辆
         List<CameraPO> cameraPOList = sensorCameraMapper.selectCameraByVehicleConfigId(vehicleConfigId);    // 摄像头
         List<OgtPO> ogtPOList = sensorOgtMapper.selectOgtByVehicleId(vehicleConfigId); // 完美传感器
         // -------------------------------- 3 发送任务消息 --------------------------------
-        manualProjectService.sendTaskMessage(manualProjectTopic, userId, projectId, maxSimulationTime, scenePOSet, vehiclePO, cameraPOList, ogtPOList);
+        manualProjectService.sendTaskMessage(projectRunningPrefix, userId, projectId, maxSimulationTime, scenePOSet, vehiclePO, cameraPOList, ogtPOList);
         // -------------------------------- 4 算法导入（一期按单机版做） --------------------------------
         String algorithmDockerImage = manualProjectService.handleAlgorithm(projectId, algorithmId);
         // -------------------------------- 5 创建 pod 开始执行 --------------------------------
         int completions = scenePOList.size();     // 结束标
-        log.info("------- ManualProjectConsumer 项目 " + projectId + " 的完成度为：" + completions);
-        log.info("------- ManualProjectConsumer 项目 " + projectId + " 的并行度为：" + parallelism);
+        log.info("ProjectConsumer--parseManualProject 项目 " + projectId + " 的完成度为：" + completions);
+        log.info("ProjectConsumer--parseManualProject 项目 " + projectId + " 的并行度为：" + parallelism);
         String jobTemplateYamlPathSource = jobTemplate + "job-template.yaml";
         String jobTemplateYamlPathTarget = jobYaml + "project-" + projectId + ".yaml";
         String yamlSource = FileUtil.read(jobTemplateYamlPathSource);
-        log.info("------- ManualProjectConsumer 模板文件为：" + yamlSource);
+        log.info("ProjectConsumer--parseManualProject 模板文件为：" + yamlSource);
         String replace0 = yamlSource.replace("job-cloud-simulation", "project-" + projectId);
         String replace1 = replace0.replace("vtd-container", "vtd-" + projectId);
         String replace2 = replace1.replace("algorithm-container", "algorithm-" + projectId);
@@ -202,7 +230,7 @@ public class ManualProjectConsumer {
         String replace6 = replace5.replace("parallelism-number", parallelism + "");
         String replace7 = replace6.replace("apiVers1on", "apiVersion");
         String replace8 = replace7.replace("1atch/v1", "batch/v1");
-        log.info("------- ManualProjectConsumer 开始执行 yaml 文件" + replace8);
+        log.info("ProjectConsumer--parseManualProject 开始执行 yaml 文件" + replace8);
         FileUtil.writeStringToLocalFile(replace8, jobTemplateYamlPathTarget);
 
 
@@ -216,7 +244,7 @@ public class ManualProjectConsumer {
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.manual-project.stop-topic}")
     @SneakyThrows
     public void stopManualProject(ConsumerRecord<String, String> stopRecord) {
-        log.info("ManualProjectConsumer--stopManualProject 接收到的项目终止消息为：" + stopRecord);
+        log.info("ProjectConsumer--stopManualProject 接收到的项目终止消息为：" + stopRecord);
         //1 读取 kafka 的项目停止信息
         /*
             {

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/manager/TaskManager.java

@@ -7,8 +7,10 @@ import com.css.simulation.resource.scheduler.mapper.IndexMapper;
 import com.css.simulation.resource.scheduler.mapper.ManualProjectMapper;
 import com.css.simulation.resource.scheduler.mapper.TaskMapper;
 import com.css.simulation.resource.scheduler.pojo.po.*;
+import com.css.simulation.resource.scheduler.pojo.to.PrefixTO;
 import com.css.simulation.resource.scheduler.pojo.to.ScoreTO;
 import com.css.simulation.resource.scheduler.util.MinioUtil;
+import com.css.simulation.resource.scheduler.util.ProjectUtil;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.minio.MinioClient;
@@ -22,6 +24,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.io.IOException;
 import java.util.*;
@@ -69,14 +72,15 @@ public class TaskManager {
     CloseableHttpClient closeableHttpClient;
     @Autowired
     RequestConfig requestConfig;
-
+    @Autowired
+    ProjectUtil projectUtil;
 
     @SneakyThrows
-    public boolean isProjectCompleted(String userId, String projectId, String taskId, String state, String podName, ClientSession session) {
-
+    @Transactional
+    public boolean isProjectCompleted(PrefixTO redisPrefix, String projectId, String taskId, String state, String podName, ClientSession session) {
         if ("Running".equals(state)) {
             // 将运行中的任务的 pod 名称放入 redis
-            stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":pod", podName);
+            stringRedisTemplate.opsForValue().set(redisPrefix.getTaskPodKey(), podName);
             taskTick(taskId); // 刷新一下心跳
             log.info("TaskManager--state 修改任务 " + taskId + "的状态为 Running，pod 名称为：" + podName);
             taskMapper.updateStateWithStartTime(taskId, state, TimeUtil.getNowForMysql());
@@ -85,11 +89,10 @@ public class TaskManager {
             String podDeleteCommand = "kubectl delete pod " + podName;
             log.info("TaskManager--state 修改任务 " + taskId + "的状态为:" + state + "，pod 名称为：" + podName + "，并执行删除 pod 命令：" + podDeleteCommand);
             if ("Aborted".equals(state)) {
-                if (retry(userId, projectId, taskId)) {
+                if (retry(projectId, taskId, redisPrefix.getTaskRetryKey(), redisPrefix.getTaskMessageKey())) {
                     taskMapper.updateStateById(DictConstants.TASK_RUNNING, taskId);
                     return false;
                 }
-                //result-path-minio: /project/manual-project/
                 String minioPathOfErrorLog = resultPathMinio + projectId + "/" + taskId + "error.log";
                 boolean objectExist = MinioUtil.isObjectExist(minioClient, bucketName, minioPathOfErrorLog);
                 String targetEvaluate;
@@ -112,7 +115,7 @@ public class TaskManager {
                 }
                 taskMapper.updateFailStateWithStopTime(taskId, state, TimeUtil.getNowForMysql(), targetEvaluate);
             } else if ("Terminated".equals(state)) {
-                if (retry(userId, projectId, taskId)) {
+                if (retry(projectId, taskId, redisPrefix.getTaskRetryKey(), redisPrefix.getTaskMessageKey())) {
                     taskMapper.updateStateById(DictConstants.TASK_RUNNING, taskId);
                     return false;
                 }
@@ -120,7 +123,7 @@ public class TaskManager {
             } else if ("PendingAnalysis".equals(state)) {
                 taskMapper.updateSuccessStateWithStopTime(taskId, state, TimeUtil.getNowForMysql());
             } else {
-                if (retry(userId, projectId, taskId)) {
+                if (retry(projectId, taskId, redisPrefix.getTaskRetryKey(), redisPrefix.getTaskMessageKey())) {
                     taskMapper.updateStateById(DictConstants.TASK_RUNNING, taskId);
                     return false;
                 }
@@ -131,25 +134,25 @@ public class TaskManager {
         int taskNum = taskMapper.selectTaskNumByProjectId(projectId);
         int endTaskNum = taskMapper.selectEndTaskNumByProjectId(projectId);    // 查询已结束的任务 'Aborted', 'PendingAnalysis', 'Terminated'
         manualProjectMapper.updateTaskCompleted(projectId, endTaskNum);
-        log.info("TaskManager--state 项目 " + projectId + " 完成进度为：" + endTaskNum + "/" + taskNum);
+        log.info("TaskManager--isProjectCompleted 项目 " + projectId + " 完成进度为：" + endTaskNum + "/" + taskNum);
         // 已结束任务数等于所有任务数量，才会准备打分；否则退出。
         return taskNum == endTaskNum;
 
     }
 
-    public boolean retry(String userId, String projectId, String taskId) {
-        log.info("TaskService--retry 重试操作收到的参数为：userId=" + userId + "，projectId=" + projectId + "，taskId=" + taskId);
+    public boolean retry(String projectId, String taskId, String taskRetryKey, String taskMessageKey) {
+        log.info("TaskService--retry 重试操作收到的参数为：projectId=" + projectId + "，taskId=" + taskId);
         //1 首先查看任务是否重试过 3 次
-        String retryString = stringRedisTemplate.opsForValue().get(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":retry");
+        String retryString = stringRedisTemplate.opsForValue().get(taskRetryKey);
         int retry = Integer.parseInt(Objects.requireNonNull(retryString));
         //2 如果重试次数没有超过 3 次，则重试
         if (retry > 3) {
             return false;
         }
-        String taskJson = stringRedisTemplate.opsForValue().get(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":message");
+        String taskJson = stringRedisTemplate.opsForValue().get(taskMessageKey);
         retry++;
         log.info("TaskService--retry 重试项目 " + projectId + " 的任务 " + taskId + "，重试次数为:" + retry + "，重新发送的消息为：" + taskJson);
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":retry", retry + "");
+        stringRedisTemplate.opsForValue().set(taskRetryKey, retry + "");
         kafkaTemplate.send(projectId, taskJson).addCallback(success -> {
             // 消息发送到的topic
             assert success != null;
@@ -158,7 +161,7 @@ public class TaskManager {
             int partition = success.getRecordMetadata().partition();
             // 消息在分区内的offset
             long offset = success.getRecordMetadata().offset();
-            log.info("------- ManualProjectConsumer 发送消息成功：\n"
+            log.info("------- ProjectConsumer 发送消息成功：\n"
                     + "主题 topic 为：" + topic + "\n"
                     + "分区 partition 为：" + partition + "\n"
                     + "偏移量为：" + offset + "\n"
@@ -169,12 +172,8 @@ public class TaskManager {
         return true;
     }
 
-    public String prepareScore(String userId, String projectId) {
-        log.info("TaskManager--prepareScore 项目 " + projectId + "准备打分！");
-        ClusterPO clusterPO = clusterMapper.selectByUserId(userId);
-        String clusterId = clusterPO.getId();
-        stringRedisTemplate.delete(manualProjectTopic + ":cluster:" + clusterId + ":running" + projectId);
-        return clusterId;
+    public void prepareScore(String projectRunningKey) {
+        stringRedisTemplate.delete(projectRunningKey);
     }
 
     @SneakyThrows
@@ -413,7 +412,8 @@ public class TaskManager {
         String projectId = taskPO.getPId();
         String userId = taskPO.getCreateUserId();
         // 刷新 redis 心跳时间
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":tick", TimeUtil.getNowString());
+        PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaksId(userId, projectId, taskId);
+        stringRedisTemplate.opsForValue().set(redisPrefix.getTaskTickKey(), TimeUtil.getNowString());
     }
 
 

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

@@ -19,4 +19,10 @@ public interface AutoSubProjectMapper {
             "from simulation_automatic_subproject\n" +
             "where id = #{subProjectId}")
     String selectCreateUserById(@Param("subProjectId") String subProjectId);
+
+
+    @Update("update simulation_automatic_subproject\n" +
+            "set now_run_state = #{nowRunState}\n" +
+            "where id = #{id}")
+    void updateNowRunStateById(@Param("nowRunState") String nowRunState, @Param("id") String id);
 }

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

@@ -20,7 +20,7 @@ public interface ManualProjectMapper {
     @Select("select id, scene, create_user_id\n" +
             "from simulation_manual_project\n" +
             "where id = #{projectId}")
-    ProjectPO selectById(@Param("projectId")String projectId);
+    ProjectPO selectById(@Param("projectId") String projectId);
 
     @ResultMap("project")
     @Select("select id, scene, create_user_id\n" +
@@ -31,22 +31,15 @@ public interface ManualProjectMapper {
 
 
     @Update("update simulation_manual_project\n" +
-            "set now_run_state  = #{state},\n" +
-            "    task_completed = '0'\n" +
+            "set now_run_state = #{nowRunState}\n" +
             "where id = #{id}")
-    void updateInit(@Param("id") String id, @Param("state") String state);
+    void updateNowRunStateById(@Param("nowRunState") String nowRunState, @Param("id") String id);
 
     @Update("update simulation_manual_project\n" +
             "set now_run_state  = #{state},\n" +
             "    finish_time = #{finishTime}\n" +
             "where id = #{id}")
-    void updateProjectState(@Param("id") String id, @Param("state") String state,@Param("finishTime") Timestamp finishTime);
-
-
-    @Update("update simulation_manual_project\n" +
-            "set task_number = #{taskNumber}, task_completed = 0\n" +
-            "where id = #{id}")
-    void updateTaskNumber(@Param("id") String id, @Param("taskNumber") int taskNumber);
+    void updateProjectState(@Param("id") String id, @Param("state") String state, @Param("finishTime") Timestamp finishTime);
 
     @Update("update simulation_manual_project\n" +
             "set task_completed = #{taskCompleted}\n" +
@@ -54,7 +47,6 @@ public interface ManualProjectMapper {
     void updateTaskCompleted(@Param("id") String id, @Param("taskCompleted") int taskCompleted);
 
 
-
     @Select("select create_user_id\n" +
             "from simulation_manual_project\n" +
             "where id = #{id}")

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

@@ -53,12 +53,6 @@ public interface TaskMapper {
 
 
 
-    @Update("update simulation_manual_project\n" +
-            "set now_run_state = #{state}\n" +
-            "where id = (select p_id from simulation_manual_project_task where id = #{taskId})")
-    void updateProjectStateByTaskId( @Param("state") String state, @Param("taskId") String taskId);
-
-
     @Update("update simulation_manual_project_task\n" +
             "set run_state = #{runState},run_start_time = #{runStartTime}\n" +
             "where id = #{id}")
@@ -105,17 +99,6 @@ public interface TaskMapper {
     void updateSuccessStateAndScoreResultWithStopTime(@Param("task") TaskPO task, @Param("runState") String runState, @Param("runStopTime") Timestamp runStopTime);
 
 
-
-    @Select("select id\n" +
-            "from simulation_manual_project_task\n" +
-            "where p_id = #{projectId} and scene_id = #{sceneId}")
-    List<String> selectIdByProjectIdAndSceneId(@Param("projectId") String projectId, @Param("sceneId") String sceneId);
-
-    @Select("select last_targer_id\n" +
-            "from simulation_manual_project_task\n" +
-            "where id = #{taskId}")
-    String selectLastTargetIdById(@Param("taskId") String taskId);
-
     @Update("update simulation_manual_project_task\n" +
             "set run_state = #{runState}\n" +
             "where id = #{taskId}")

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/pojo/to/InfoTO.java

@@ -14,5 +14,5 @@ public class InfoTO {
     private String project_id;
     private String task_id;
     private String task_path;
-    private Integer default_time;
+    private Long default_time;
 }

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/pojo/to/PrefixTO.java

@@ -0,0 +1,25 @@
+package com.css.simulation.resource.scheduler.pojo.to;
+
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class PrefixTO {
+
+    private String clusterPrefix;
+    private String clusterRunningPrefix;
+    private String clusterWaitingPrefix;
+    private String projectRunningKey;
+    private String projectWaitingKey;
+    private String projectCheckKey;
+    private String taskTickKey;
+    private String taskPodKey;
+    private String taskRetryKey;
+    private String taskMessageKey;
+}

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

@@ -3,14 +3,17 @@ package com.css.simulation.resource.scheduler.scheduler;
 import api.common.pojo.constants.DictConstants;
 import api.common.pojo.dto.ProjectMessageDTO;
 import api.common.util.*;
-import com.css.simulation.resource.scheduler.consumer.ManualProjectConsumer;
+import com.css.simulation.resource.scheduler.consumer.ProjectConsumer;
 import com.css.simulation.resource.scheduler.mapper.ClusterMapper;
 import com.css.simulation.resource.scheduler.mapper.ManualProjectMapper;
 import com.css.simulation.resource.scheduler.mapper.TaskMapper;
+import com.css.simulation.resource.scheduler.mapper.UserMapper;
 import com.css.simulation.resource.scheduler.pojo.po.ClusterPO;
 import com.css.simulation.resource.scheduler.pojo.po.ProjectPO;
 import com.css.simulation.resource.scheduler.pojo.po.TaskPO;
+import com.css.simulation.resource.scheduler.pojo.to.PrefixTO;
 import com.css.simulation.resource.scheduler.service.TaskService;
+import com.css.simulation.resource.scheduler.util.ProjectUtil;
 import io.kubernetes.client.openapi.ApiClient;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -56,7 +59,11 @@ public class ProjectScheduler {
     @Autowired
     KafkaTemplate<String, String> kafkaTemplate;
     @Autowired
-    ManualProjectConsumer manualProjectConsumer;
+    ProjectConsumer projectConsumer;
+    @Autowired
+    UserMapper userMapper;
+    @Autowired
+    ProjectUtil projectUtil;
 
 
     /**
@@ -65,6 +72,7 @@ public class ProjectScheduler {
     @Scheduled(fixedDelay = 60 * 1000)
     @SneakyThrows
     public void dispatchProject() {
+
         //1 查询已经排队的项目，即已经在页面上点击运行
         List<ProjectPO> projectList = manualProjectMapper.selectByNowRunState(DictConstants.PROJECT_RUNNING);
         for (ProjectPO project : projectList) {
@@ -76,51 +84,48 @@ public class ProjectScheduler {
                 return;
             }
             String clusterId = clusterPO.getId();
-            if (StringUtil.isNotEmpty(redisTemplate.opsForValue().get(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId))) {
+            PrefixTO redisPrefix = projectUtil.getRedisPrefixByClusterIdAndProjectId(clusterId, projectId);
+            if (StringUtil.isNotEmpty(redisTemplate.opsForValue().get(redisPrefix.getProjectRunningKey()))) {
                 continue; // 判断项目是否已经在执行，如果执行则 continue
             }
             int simulationLicenseNumber = clusterPO.getNumSimulationLicense();
             // 获取该用户正在运行的项目数量
-            Set<String> runningProjectSet = redisTemplate.keys(manualProjectTopic + ":cluster:" + clusterId + ":running:" + "*");
+            Set<String> runningProjectSet = redisTemplate.keys(redisPrefix.getClusterRunningPrefix() + "*");
             if (CollectionUtil.isNotEmpty(runningProjectSet)) {
                 long parallelismSum = 0;
                 for (String runningProjectKey : runningProjectSet) {
-                    String runningProjectJsonTemp = redisTemplate.opsForValue().get(runningProjectKey);
-                    ProjectMessageDTO runningProjectMessageTemp = JsonUtil.jsonToBean(runningProjectJsonTemp, ProjectMessageDTO.class);
-                    parallelismSum += runningProjectMessageTemp.getParallelism();
+                    parallelismSum += JsonUtil.jsonToBean(redisTemplate.opsForValue().get(runningProjectKey), ProjectMessageDTO.class).getParallelism();
                 }
                 if (parallelismSum < simulationLicenseNumber) {
-                    Set<String> waitingProjectSet = redisTemplate.keys(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + "*");
+                    Set<String> waitingProjectSet = redisTemplate.keys(redisPrefix.getClusterWaitingPrefix() + "*");
                     if (CollectionUtil.isEmpty(waitingProjectSet)) {
                         return;
                     }
                     for (String waitingProjectKey : waitingProjectSet) {
-                        String waitingProjectJsonTemp = redisTemplate.opsForValue().get(waitingProjectKey);
-                        ProjectMessageDTO waitingProjectMessageTemp = JsonUtil.jsonToBean(waitingProjectJsonTemp, ProjectMessageDTO.class);
-                        int parallelism = waitingProjectMessageTemp.getParallelism();
+                        Long parallelism = JsonUtil.jsonToBean(redisTemplate.opsForValue().get(waitingProjectKey), ProjectMessageDTO.class).getParallelism();
                         if (parallelismSum + parallelism < simulationLicenseNumber) {
-                            String projectJson = redisTemplate.opsForValue().get(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + projectId);
-                            redisTemplate.delete(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + projectId);
-                            assert projectJson != null;
-                            redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId, projectJson);
-                            log.info("ProjectScheduler--dispatchProject 项目 " + projectId + " 从等待队列进入执行状态！");
-                            manualProjectConsumer.parseManualProject(projectJson);
+                            run(clusterId, projectId, redisPrefix.getProjectWaitingKey(), redisPrefix.getProjectRunningKey());
                         }
                     }
                 }
             } else {
-                String projectJson = redisTemplate.opsForValue().get(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + projectId);
-                redisTemplate.delete(manualProjectTopic + ":cluster:" + clusterId + ":waiting:" + projectId);
-                assert projectJson != null;
-                redisTemplate.opsForValue().set(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId, projectJson);
-                log.info("ProjectScheduler--dispatchProject 项目 " + projectId + " 从等待队列进入执行状态！");
-                manualProjectConsumer.parseManualProject(projectJson);
+                run(clusterId, projectId, redisPrefix.getProjectWaitingKey(), redisPrefix.getProjectRunningKey());
             }
 
         }
 
     }
 
+    public void run(String clusterId, String projectId, String projectWaitingKey, String projectRunningKey) {
+        String clusterPrefix = "cluster:" + clusterId;
+        String projectJson = redisTemplate.opsForValue().get(projectWaitingKey);
+        redisTemplate.delete(projectWaitingKey);
+        assert projectJson != null;
+        redisTemplate.opsForValue().set(projectRunningKey, projectJson);
+        log.info("ProjectScheduler--run 项目 " + projectId + " 从等待队列进入执行状态！");
+        projectConsumer.parseManualProject(projectJson, clusterPrefix, projectRunningKey);
+    }
+
 
     /**
      * 处理 pod 超时
@@ -130,26 +135,27 @@ public class ProjectScheduler {
      */
     @Scheduled(fixedDelay = 60 * 1000)
     public void taskTimeout() throws IOException {
-
         long timeout = 2 * 60 * 1000L;
-
         SshClient client = SshUtil.getClient();
         ClientSession session = SshUtil.getSession(client, hostname, username, password);
         List<TaskPO> executingTaskList = taskMapper.selectByRunState(DictConstants.TASK_RUNNING);
         log.info("ProjectScheduler--taskTimeout 正在运行的任务有：" + executingTaskList);
         if (executingTaskList != null && executingTaskList.size() > 0) {
             for (TaskPO task : executingTaskList) {
-                String taskId = task.getId();
-                String projectId = task.getPId();
                 String userId = task.getCreateUserId();
-                String tickTime = redisTemplate.opsForValue().get(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":tick");
+                String projectId = task.getPId();
+                String taskId = task.getId();
+                PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaksId(userId, projectId, taskId);
+                // 获取心跳时间
+                String tickTime = redisTemplate.opsForValue().get(redisPrefix.getTaskTickKey());
                 if (StringUtil.isEmpty(tickTime)) {
-                    log.error(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":tick" + "，该 key 的心跳时间为空！");
+                    log.error(redisPrefix.getTaskTickKey() + "，该 key 的心跳时间为空！");
                     continue;
                 }
                 long lastTickTime = Long.parseLong(tickTime);
+                // 如果心跳超时则更改任务状态为 Aborted
                 if (TimeUtil.getNow() - lastTickTime > timeout) {
-                    String podName = redisTemplate.opsForValue().get(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":pod");
+                    String podName = redisTemplate.opsForValue().get(redisPrefix.getTaskPodKey());
                     taskService.taskState(taskId, DictConstants.TASK_ABORTED, podName);
                 }
             }
@@ -163,7 +169,8 @@ public class ProjectScheduler {
      * 检查如果有 job 在运行但是 pod 全部关闭的情况，此时需要重启一下 job
      */
     @Scheduled(fixedDelay = 30 * 1000)
-    public void projectCheck() throws IOException {
+    @SneakyThrows
+    public void projectCheck() {
         SshClient client = SshUtil.getClient();
         ClientSession session = SshUtil.getSession(client, hostname, username, password);
 
@@ -171,48 +178,43 @@ public class ProjectScheduler {
         List<ProjectPO> projectIdList = manualProjectMapper.selectByNowRunState(DictConstants.PROJECT_RUNNING);
         log.info("ProjectScheduler--projectCheck 运行中的项目有：" + projectIdList);
         //2 根据 projectId 获取 pod
-        projectIdList.forEach(project -> {
+        for (ProjectPO project : projectIdList) {
             String projectId = project.getId();
             String userId = project.getCreateUserId();
-            try {
-                String checkKey = manualProjectTopic + ":" + userId + ":" + projectId + ":check";
-                String lastNowString = redisTemplate.opsForValue().get(checkKey);
-                String podList = SshUtil.execute(session, "kubectl get pod | grep project-" + projectId);
-                log.info("ProjectScheduler--projectCheck 项目 " + projectId + " 正在运行的 pod 为：\n" + podList);
-                int taskNumber = StringUtil.countSubString(podList, "project");
-                if (StringUtil.isEmpty(lastNowString) && taskNumber == 0) {
+            PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectId(userId, projectId);
+            String checkKey = redisPrefix.getProjectCheckKey();
+            String lastNowString = redisTemplate.opsForValue().get(checkKey);
+            String podList = SshUtil.execute(session, "kubectl get pod | grep project-" + projectId);
+            log.info("ProjectScheduler--projectCheck 项目 " + projectId + " 正在运行的 pod 为：\n" + podList);
+            int taskNumber = StringUtil.countSubString(podList, "project");
+            if (StringUtil.isEmpty(lastNowString) && taskNumber == 0) {  // 为空代表第一次，先设置时间
+                redisTemplate.opsForValue().set(checkKey, TimeUtil.getNowString());
+            }
+            if (StringUtil.isNotEmpty(lastNowString) && taskNumber == 0) {  // 非空则开始检查
+                // 判断两次是否超过2分钟
+                //3 如果 pod 为空，则重启 job
+                long lastNow = Long.parseLong(lastNowString);
+                long now = Long.parseLong(TimeUtil.getNowString());
+                if (now - lastNow > (long) 120 * 1000) {
                     redisTemplate.opsForValue().set(checkKey, TimeUtil.getNowString());
-                }
-                if (StringUtil.isNotEmpty(lastNowString) && taskNumber == 0) {
-                    // 判断两次是否超过2分钟
-                    //3 如果 pod 为空，则重启 job
-                    long lastNow = Long.parseLong(lastNowString);
-                    long now = Long.parseLong(TimeUtil.getNowString());
-
-                    if (now - lastNow > (long) 120 * 1000) {
-                        redisTemplate.opsForValue().set(checkKey, TimeUtil.getNowString());
-                        SshUtil.execute(session, "kubectl delete job project-" + projectId);
-                        Thread.sleep(15000);
-                        while (true) {
-                            log.info("ProjectScheduler--projectCheck 准备重启项目 " + projectId);
-                            String podList2 = SshUtil.execute(session, "kubectl get pod | grep project-" + projectId);
-                            log.info("ProjectScheduler--projectCheck 项目 " + projectId + " 剩余的 pod 信息为：\n" + podList2);
-                            int taskNumber2 = StringUtil.countSubString(podList2, "project");
-                            if (taskNumber2 == 0) {
-                                break;
-                            }
+                    SshUtil.execute(session, "kubectl delete job project-" + projectId);
+                    Thread.sleep(15000);
+                    while (true) {
+                        log.info("ProjectScheduler--projectCheck 准备重启项目 " + projectId);
+                        String podList2 = SshUtil.execute(session, "kubectl get pod | grep project-" + projectId);
+                        log.info("ProjectScheduler--projectCheck 项目 " + projectId + " 剩余的 pod 信息为：\n" + podList2);
+                        int taskNumber2 = StringUtil.countSubString(podList2, "project");
+                        if (taskNumber2 == 0) {
+                            break;
                         }
-                        Thread.sleep(15000);
-                        log.info("ProjectScheduler--projectCheck 重新执行项目" + projectId);
-                        String jobTemplateYamlPathTarget = jobYaml + "project-" + projectId + ".yaml";
-                        SshUtil.execute(session, "kubectl apply -f " + jobTemplateYamlPathTarget);
                     }
+                    Thread.sleep(15000);
+                    log.info("ProjectScheduler--projectCheck 重新执行项目" + projectId);
+                    String jobTemplateYamlPathTarget = jobYaml + "project-" + projectId + ".yaml";
+                    SshUtil.execute(session, "kubectl apply -f " + jobTemplateYamlPathTarget);
                 }
-            } catch (IOException | InterruptedException e) {
-                e.printStackTrace();
             }
-        });
-
+        }
         session.close();
         client.stop();
 

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

@@ -14,8 +14,6 @@ import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.ibatis.session.ExecutorType;
-import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -57,6 +55,8 @@ public class ManualProjectService {
     @Autowired
     ManualProjectMapper manualProjectMapper;
     @Autowired
+    AutoSubProjectMapper autoSubProjectMapper;
+    @Autowired
     TaskMapper taskMapper;
     @Autowired
     IndexTemplateMapper indexTemplateMapper;
@@ -77,16 +77,24 @@ public class ManualProjectService {
     // -------------------------------- Comment --------------------------------
 
     @Transactional
-    public void prepare(String manualProjectTopic, String userId, String projectId, String projectJson) {
-        //1 redis 设置项目已完成任务为 0
-        Set<String> oldKeys = stringRedisTemplate.keys(manualProjectTopic + ":" + projectId + "*");
+    public void prepare(String clusterPrefix, String projectId, String projectType) {
+        //1 如果集群中有该项目旧的信息则直接删除
+        Set<String> oldKeys = stringRedisTemplate.keys(clusterPrefix + "*");
         if (CollectionUtil.isNotEmpty(oldKeys)) {
-            stringRedisTemplate.delete(oldKeys);
+            for (String oldKey : oldKeys) {
+                if (oldKey.contains(projectId)) {
+                    stringRedisTemplate.delete(oldKeys);
+                }
+            }
         }
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":completed", "0");
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":start", projectJson);
-        manualProjectMapper.updateInit(projectId, DictConstants.PROJECT_RUNNING);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 。
-        // 将该 project 下所有旧的任务和指标得分删除。
+        //2 将项目状态修改为执行中
+        if (DictConstants.PROJECT_TYPE_MANUAL.equals(projectType)) {
+            manualProjectMapper.updateNowRunStateById(DictConstants.PROJECT_RUNNING, projectId);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 。
+        } else if (DictConstants.PROJECT_TYPE_AUTO_SUB.equals(projectType)) {
+            autoSubProjectMapper.updateNowRunStateById(DictConstants.PROJECT_RUNNING, projectId);
+        }
+
+        //3 将该 project 下所有旧的任务和指标得分删除。
         taskMapper.deleteByProject(projectId);
         indexMapper.deleteFirstTargetScoreByProjectId(projectId);
         indexMapper.deleteLastTargetScoreByProjectId(projectId);
@@ -94,15 +102,17 @@ public class ManualProjectService {
 
     @SneakyThrows
     @Transactional
-    public List<ScenePO> handlePackage(String manualProjectTopic, String userId, String projectId, String packageId) {
+    public List<ScenePO> handlePackage(String projectRunningPrefix, String projectId, String packageId) {
+        String allIndexPrefix = projectRunningPrefix + ":package:" + packageId + ":all";
+        String leafIndexPrefix = projectRunningPrefix + ":package:" + packageId + ":leaf";
 
         //1 查询该场景包的所有指标列表存入 redis，包删了无所谓，但要过滤删掉的指标
         List<IndexTemplatePO> allIndexList = indexTemplateMapper.selectByPackageIdIncludeDeleted(packageId);
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":package:" + packageId + ":all", JsonUtil.listToJson(allIndexList));
+        stringRedisTemplate.opsForValue().set(allIndexPrefix, JsonUtil.listToJson(allIndexList));
 
         //2 查询场景包叶子指标
         List<IndexTemplatePO> leafIndexList = allIndexList.stream().filter(index -> StringUtil.isNotEmpty(index.getRuleId())).collect(Collectors.toList());
-        stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":package:" + packageId + ":leaf", JsonUtil.listToJson(leafIndexList));
+        stringRedisTemplate.opsForValue().set(leafIndexPrefix, JsonUtil.listToJson(leafIndexList));
         log.info("ManualProjectService--handlePackage 项目 " + projectId + " 的叶子指标为：" + leafIndexList);
         List<ScenePO> sceneList = new ArrayList<>();
         leafIndexList.forEach(leafIndex -> {
@@ -122,125 +132,115 @@ public class ManualProjectService {
                 sceneList.addAll(sceneMapper.selectAccidentByIdList(accidentIdList));
             }
         });
-        manualProjectMapper.updateTaskNumber(projectId, sceneList.size());
-        log.info("ManualProjectService--handlePackage 项目" + projectId + " 共有 " + sceneList.size() + " 个任务，对应 " + sceneList.size() + " 个场景！");
+        log.info("ManualProjectService--handlePackage 项目" + projectId + " 共有 " + sceneList.size() + " 个任务！");
         return sceneList;
     }
 
 
-    public void sendTaskMessage(String manualProjectTopic, String userId, String projectId, int maxSimulationTime, Set<ScenePO> scenePOSet, VehiclePO vehiclePO, List<CameraPO> cameraPOList, List<OgtPO> ogtPOList) {
+    public void sendTaskMessage(String projectRunningPrefix, String userId, String projectId, Long maxSimulationTime, Set<ScenePO> scenePOSet, VehiclePO vehiclePO, List<CameraPO> cameraPOList, List<OgtPO> ogtPOList) {
+        final int[] messageNumber = {0};
+        for (ScenePO scenePO : scenePOSet) {
+            String sceneId = scenePO.getId();
+            //3-1 可能会存在多个指标下有同样的场景，所以会查出多个指标，多个指标的场景需要发送多次
+            List<String> lastTargetIdList = indexMapper.selectLeafIndexIdByProjectAndSceneId(projectId, "%" + sceneId + "%");
+            lastTargetIdList.forEach(lastTargetId -> {
+                String taskId = StringUtil.getRandomUUID();
+                String taskRetryPrefix = projectRunningPrefix + ":task:" + taskId + ":retry";
+                String taskMessagePrefix = projectRunningPrefix + ":task:" + taskId + ":message";
+                // 设置任务重试次数为 0，方便任务进行最大3次的重试。
+                stringRedisTemplate.opsForValue().set(taskRetryPrefix, "0");
+                // 保存任务信息
+                TaskPO taskPO = TaskPO.builder() // run_start_time 和 run_end_time 不填
+                        .id(taskId)
+                        .pId(projectId)
+                        .sceneId(sceneId)
+                        .lastTargetId(lastTargetId)
+                        .sceneName(scenePO.getName())
+                        .sceneType(scenePO.getType())
+                        .runState(DictConstants.TASK_PENDING)
+                        .runResultFilePath(resultPathMinio + projectId + "/" + taskId)
+                        .build();
+                taskPO.setCreateTime(TimeUtil.getNowForMysql());
+                taskPO.setCreateUserId(userId);
+                taskPO.setModifyTime(TimeUtil.getNowForMysql());
+                taskPO.setModifyUserId(userId);
+                taskPO.setModifyTime(TimeUtil.getNowForMysql());
+                taskPO.setIsDeleted("0");
+                taskMapper.insert(taskPO);
 
-        try (SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) {
-            IndexMapper indexMapper = sqlSession.getMapper(IndexMapper.class);
-            TaskMapper taskMapper = sqlSession.getMapper(TaskMapper.class);
+                // 组装 task 消息
+                TaskTO taskTO = TaskTO.builder()
+                        .info(InfoTO.builder()
+                                .project_id(taskPO.getPId())
+                                .task_id(taskPO.getId())
+                                .task_path(taskPO.getRunResultFilePath())
+                                .default_time(maxSimulationTime)
+                                .build())
+                        .scenario(ScenarioTO.builder()
+                                .scenario_osc(scenePO.getScenarioOsc())
+                                .scenario_odr(scenePO.getScenarioOdr())
+                                .scenario_osgb(scenePO.getScenarioOsgb())
+                                .build())
+                        .vehicle(VehicleTO.builder()
+                                .model(ModelTO.builder()
+                                        .model_label(vehiclePO.getModelLabel())
+                                        .build())
+                                .dynamics(DynamicsTO.builder()
+                                        .dynamics_maxspeed(vehiclePO.getMaxSpeed())
+                                        .dynamics_enginepower(vehiclePO.getEnginePower())
+                                        .dynamics_maxdecel(vehiclePO.getMaxDeceleration())
+                                        .dynamics_maxsteering(vehiclePO.getMaxSteeringAngle())
+                                        .dynamics_mass(vehiclePO.getMass())
+                                        .dynamics_frontsurfaceeffective(vehiclePO.getFrontSurfaceEffective())
+                                        .dynamics_airdragcoefficient(vehiclePO.getAirDragCoefficient())
+                                        .dynamics_rollingresistance(vehiclePO.getRollingResistanceCoefficient())
+                                        .dynamics_wheeldiameter(vehiclePO.getWheelDiameter())
+                                        .dynamics_wheeldrive(vehiclePO.getWheelDrive())
+                                        .dynamics_overallefficiency(vehiclePO.getOverallEfficiency())
+                                        .dynamics_distfront(vehiclePO.getFrontDistance())
+                                        .dynamics_distrear(vehiclePO.getRearDistance())
+                                        .dynamics_distleft(vehiclePO.getLeftDistance())
+                                        .dynamics_distright(vehiclePO.getRightDistance())
+                                        .dynamics_distheight(vehiclePO.getHeightDistance())
+                                        .dynamics_wheelbase(vehiclePO.getWheelbase())
+                                        .build())
+                                .sensors(SensorsTO.builder()   // 根据 vehicleId 查询绑定的传感器列表
+                                        .camera(cameraPOList)
+                                        .OGT(ogtPOList)
+                                        .build())
+                                .build())
+                        .build();
 
-            final int[] messageNumber = {0};
-            scenePOSet.forEach(scenePO -> {
-                String sceneId = scenePO.getId();
-                //3-1 可能会存在多个指标下有同样的场景，所以会查出多个指标，多个指标的场景需要发送多次
-                List<String> lastTargetIdList = indexMapper.selectLeafIndexIdByProjectAndSceneId(projectId, "%" + sceneId + "%");
-                lastTargetIdList.forEach(lastTargetId -> {
-                    String taskId = StringUtil.getRandomUUID();
-                    // 设置任务重试次数为 0，方便任务进行最大3次的重试。
-                    stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":retry", "0");
-                    // 保存任务信息
-                    TaskPO taskPO = TaskPO.builder() // run_start_time 和 run_end_time 不填
-                            .id(taskId)
-                            .pId(projectId)
-                            .sceneId(sceneId)
-                            .lastTargetId(lastTargetId)
-                            .sceneName(scenePO.getName())
-                            .sceneType(scenePO.getType())
-                            .runState(DictConstants.TASK_PENDING)
-                            .runResultFilePath(resultPathMinio + projectId + "/" + taskId)
-                            .build();
-                    taskPO.setCreateTime(TimeUtil.getNowForMysql());
-                    taskPO.setCreateUserId(userId);
-                    taskPO.setModifyTime(TimeUtil.getNowForMysql());
-                    taskPO.setModifyUserId(userId);
-                    taskPO.setModifyTime(TimeUtil.getNowForMysql());
-                    taskPO.setIsDeleted("0");
-                    taskMapper.insert(taskPO);
-
-                    // 组装 task 消息
-                    TaskTO taskTO = TaskTO.builder()
-                            .info(InfoTO.builder()
-                                    .project_id(taskPO.getPId())
-                                    .task_id(taskPO.getId())
-                                    .task_path(taskPO.getRunResultFilePath())
-                                    .default_time(maxSimulationTime)
-                                    .build())
-                            .scenario(ScenarioTO.builder()
-                                    .scenario_osc(scenePO.getScenarioOsc())
-                                    .scenario_odr(scenePO.getScenarioOdr())
-                                    .scenario_osgb(scenePO.getScenarioOsgb())
-                                    .build())
-                            .vehicle(VehicleTO.builder()
-                                    .model(ModelTO.builder()
-                                            .model_label(vehiclePO.getModelLabel())
-                                            .build())
-                                    .dynamics(DynamicsTO.builder()
-                                            .dynamics_maxspeed(vehiclePO.getMaxSpeed())
-                                            .dynamics_enginepower(vehiclePO.getEnginePower())
-                                            .dynamics_maxdecel(vehiclePO.getMaxDeceleration())
-                                            .dynamics_maxsteering(vehiclePO.getMaxSteeringAngle())
-                                            .dynamics_mass(vehiclePO.getMass())
-                                            .dynamics_frontsurfaceeffective(vehiclePO.getFrontSurfaceEffective())
-                                            .dynamics_airdragcoefficient(vehiclePO.getAirDragCoefficient())
-                                            .dynamics_rollingresistance(vehiclePO.getRollingResistanceCoefficient())
-                                            .dynamics_wheeldiameter(vehiclePO.getWheelDiameter())
-                                            .dynamics_wheeldrive(vehiclePO.getWheelDrive())
-                                            .dynamics_overallefficiency(vehiclePO.getOverallEfficiency())
-                                            .dynamics_distfront(vehiclePO.getFrontDistance())
-                                            .dynamics_distrear(vehiclePO.getRearDistance())
-                                            .dynamics_distleft(vehiclePO.getLeftDistance())
-                                            .dynamics_distright(vehiclePO.getRightDistance())
-                                            .dynamics_distheight(vehiclePO.getHeightDistance())
-                                            .dynamics_wheelbase(vehiclePO.getWheelbase())
-                                            .build())
-                                    .sensors(SensorsTO.builder()   // 根据 vehicleId 查询绑定的传感器列表
-                                            .camera(cameraPOList)
-                                            .OGT(ogtPOList)
-                                            .build())
-                                    .build())
-                            .build();
-
-                    //4-4 将对象转成 json
-                    String taskJson = "";
-                    try {
-                        taskJson = JsonUtil.beanToJson(taskTO);
-                    } catch (JsonProcessingException e) {
-                        e.printStackTrace();
-                    }
-                    //4-5 将 projectId 作为 topic 名称，发送 task 信息到 kafka
-                    String finalTaskJson = taskJson;
-                    kafkaTemplate.send(projectId, taskJson).addCallback(success -> {
-                        // 消息发送到的topic
-                        assert success != null;
-                        String topic = success.getRecordMetadata().topic();
-                        // 消息发送到的分区
-                        int partition = success.getRecordMetadata().partition();
-                        // 消息在分区内的offset
-                        long offset = success.getRecordMetadata().offset();
-                        log.info("------- ManualProjectConsumer 发送消息成功：\n"
-                                + "主题 topic 为：" + topic + "\n"
-                                + "分区 partition 为：" + partition + "\n"
-                                + "偏移量为：" + offset + "\n"
-                                + "消息体为：" + finalTaskJson);
-                    }, failure -> {
-                        log.error("------- 发送消息失败：" + failure.getMessage());
-                    });
-                    messageNumber[0] = messageNumber[0] + 1;
-                    stringRedisTemplate.opsForValue().set(manualProjectTopic + ":" + userId + ":" + projectId + ":task:" + taskId + ":message", taskJson);
+                //4-4 将对象转成 json
+                String taskJson = "";
+                try {
+                    taskJson = JsonUtil.beanToJson(taskTO);
+                } catch (JsonProcessingException e) {
+                    e.printStackTrace();
+                }
+                //4-5 将 projectId 作为 topic 名称，发送 task 信息到 kafka
+                String finalTaskJson = taskJson;
+                stringRedisTemplate.opsForValue().set(taskMessagePrefix, finalTaskJson);
+                kafkaTemplate.send(projectId, taskJson).addCallback(success -> {
+                    // 消息发送到的topic
+                    assert success != null;
+                    String topic = success.getRecordMetadata().topic();
+                    // 消息发送到的分区
+                    int partition = success.getRecordMetadata().partition();
+                    // 消息在分区内的offset
+                    long offset = success.getRecordMetadata().offset();
+                    log.info("------- ProjectConsumer 发送消息成功：\n"
+                            + "主题 topic 为：" + topic + "\n"
+                            + "分区 partition 为：" + partition + "\n"
+                            + "偏移量为：" + offset + "\n"
+                            + "消息体为：" + finalTaskJson);
+                }, failure -> {
+                    log.error("------- 发送消息失败：" + failure.getMessage());
                 });
+                messageNumber[0] = messageNumber[0] + 1;
             });
-            log.info("------- ManualProjectConsumer 共发送了" + messageNumber[0] + " 条消息！");
-            sqlSession.commit();
-        } catch (Exception e) {
-            throw new RuntimeException("ManualProjectService--sendTaskMessage 发送任务消息出错：" + e.getMessage());
         }
-
-
+        log.info("ProjectConsumer 共发送了" + messageNumber[0] + " 条消息！");
     }
 
 

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

@@ -8,6 +8,8 @@ import com.css.simulation.resource.scheduler.mapper.IndexTemplateMapper;
 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.pojo.to.PrefixTO;
+import com.css.simulation.resource.scheduler.util.ProjectUtil;
 import io.minio.MinioClient;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -47,6 +49,8 @@ public class TaskService {
     TaskMapper taskMapper;
     @Autowired
     ManualProjectMapper manualProjectMapper;
+    @Autowired
+    ProjectUtil projectUtil;
 
 
     @SneakyThrows
@@ -59,10 +63,11 @@ public class TaskService {
         }
         String projectId = taskPO.getPId();
         String userId = taskPO.getCreateUserId();
+        PrefixTO redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaksId(userId, projectId, taskId);
         SshClient client = SshUtil.getClient();
         ClientSession session = SshUtil.getSession(client, hostname, username, password);
         //1 判断项目是否已完成
-        boolean projectCompleted = taskManager.isProjectCompleted(userId, projectId, taskId, state, podName, session);
+        boolean projectCompleted = taskManager.isProjectCompleted(redisPrefix, projectId, taskId, state, podName, session);
         if (!projectCompleted) {
             session.close();
             client.stop();
@@ -70,16 +75,17 @@ public class TaskService {
         }
 
         //2 准备打分
-        String clusterId = taskManager.prepareScore(userId, projectId);
+        log.info("TaskService--taskState 项目 " + projectId + "准备打分！");
+        taskManager.prepareScore(redisPrefix.getProjectRunningKey());
 
         //3 打分
-        taskManager.score(userId, projectId,session);
+        taskManager.score(userId, projectId, session);
 
         // -------------------------------- 收尾 --------------------------------
         manualProjectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED, TimeUtil.getNowForMysql());   // 修改该 project 的状态为已完成
         log.info("TaskManager--score 项目 " + projectId + " 执行完成！");
 
-        stringRedisTemplate.delete(manualProjectTopic + ":cluster:" + clusterId + ":running:" + projectId);
+        stringRedisTemplate.delete(redisPrefix.getProjectRunningKey());
         // 删除所有 key
 //        Set<String> keys = redisTemplate.keys("manualProject:" + projectId + "*");
 //        assert keys != null;

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/util/ProjectUtil.java

@@ -0,0 +1,125 @@
+package com.css.simulation.resource.scheduler.util;
+
+import api.common.pojo.constants.DictConstants;
+import com.css.simulation.resource.scheduler.mapper.ClusterMapper;
+import com.css.simulation.resource.scheduler.mapper.UserMapper;
+import com.css.simulation.resource.scheduler.pojo.po.UserPO;
+import com.css.simulation.resource.scheduler.pojo.to.PrefixTO;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+/**
+ * 任务结果打分
+ */
+@Component
+@Slf4j
+public class ProjectUtil {
+
+    @Autowired
+    UserMapper userMapper;
+    @Autowired
+    ClusterMapper clusterMapper;
+
+
+    public PrefixTO getRedisPrefixByUserIdAndProjectIdAndTaksId(String userId, String projectId, String taskId) {
+        //3 获取用户类型（管理员账户、管理员子账户、普通账户、普通子账户）（独占、共享）
+        UserPO userPO = userMapper.selectById(userId);
+        String roleCode = userPO.getRoleCode();
+        String useType = userPO.getUseType();
+        String clusterId;
+        if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
+            clusterId = DictConstants.SYSTEM_CLUSTER_ID;
+        } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
+            clusterId = clusterMapper.selectByUserId(userId).getId();
+        } else if (DictConstants.ROLE_CODE_SUBUESR.equals(roleCode)) {
+            if (DictConstants.USE_TYPE_EXCLUSIVE.equals(useType)) {   //3-3 普通子账户，根据自己的独占节点排队
+                clusterId = clusterMapper.selectByUserId(userId).getId();
+            } else {    //3-4 共享子账户，根据父账户的共享节点排队
+                String parentUserId = userPO.getCreateUserId();
+                clusterId = clusterMapper.selectByUserId(parentUserId).getId();
+            }
+        } else {
+            throw new RuntimeException("ProjectUtil--getRedisPrefixByUserIdAndProjectIdAndTaksId 未知账户类型，无法获取集群信息！");
+        }
+        String clusterPrefix = "cluster:" + clusterId;
+        String clusterRunningPrefix = clusterPrefix + ":running";
+        String projectRunningKey = clusterRunningPrefix + ":" + projectId;
+        String taskTickKey = projectRunningKey + ":task:" + taskId + ":tick";
+        String taskPodKey = projectRunningKey + ":task:" + taskId + ":pod";
+        String taskRetryKey = projectRunningKey + ":task:" + taskId + ":retry";
+        String taskMessageKey = projectRunningKey + ":task:" + taskId + ":message";
+
+        return PrefixTO.builder()
+                .clusterPrefix(clusterPrefix)
+                .clusterRunningPrefix(clusterRunningPrefix)
+                .clusterWaitingPrefix(clusterPrefix)
+                .taskTickKey(taskTickKey)
+                .taskPodKey(taskPodKey)
+                .taskRetryKey(taskRetryKey)
+                .taskMessageKey(taskMessageKey)
+                .build();
+
+    }
+
+
+    public PrefixTO getRedisPrefixByUserIdAndProjectId(String userId, String projectId) {
+        //3 获取用户类型（管理员账户、管理员子账户、普通账户、普通子账户）（独占、共享）
+        UserPO userPO = userMapper.selectById(userId);
+        String roleCode = userPO.getRoleCode();
+        String useType = userPO.getUseType();
+        String clusterId;
+        if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
+            clusterId = DictConstants.SYSTEM_CLUSTER_ID;
+        } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
+            clusterId = clusterMapper.selectByUserId(userId).getId();
+        } else if (DictConstants.ROLE_CODE_SUBUESR.equals(roleCode)) {
+            if (DictConstants.USE_TYPE_EXCLUSIVE.equals(useType)) {   //3-3 普通子账户，根据自己的独占节点排队
+                clusterId = clusterMapper.selectByUserId(userId).getId();
+            } else {    //3-4 共享子账户，根据父账户的共享节点排队
+                String parentUserId = userPO.getCreateUserId();
+                clusterId = clusterMapper.selectByUserId(parentUserId).getId();
+            }
+        } else {
+            throw new RuntimeException("ProjectUtil--getRedisPrefixByUserIdAndProjectIdAndTaksId 未知账户类型，无法获取集群信息！");
+        }
+        String clusterPrefix = "cluster:" + clusterId;
+        String clusterRunningPrefix = clusterPrefix + ":running";
+        String clusterWaitingPrefix = clusterPrefix + ":waiting";
+        String projectRunningKey = clusterRunningPrefix + ":" + projectId;
+        String projectWaitingKey = clusterWaitingPrefix + ":" + projectId;
+        String projectCheckKey = clusterWaitingPrefix + ":" + projectId + ":check";
+
+        return PrefixTO.builder()
+                .clusterPrefix(clusterPrefix)
+                .clusterRunningPrefix(clusterRunningPrefix)
+                .clusterWaitingPrefix(clusterWaitingPrefix)
+                .projectRunningKey(projectRunningKey)
+                .projectWaitingKey(projectWaitingKey)
+                .projectCheckKey(projectCheckKey)
+                .build();
+
+    }
+
+
+    public PrefixTO getRedisPrefixByClusterIdAndProjectId(String clusterId, String projectId) {
+        String clusterPrefix = "cluster:" + clusterId;
+        String clusterRunningPrefix = clusterPrefix + ":running";
+        String clusterWaitingPrefix = clusterPrefix + ":waiting";
+        String projectRunningKey = clusterRunningPrefix + ":" + projectId;
+        String projectWaitingKey = clusterWaitingPrefix + ":" + projectId;
+        String projectCheckKey = clusterWaitingPrefix + ":" + projectId + ":check";
+
+        return PrefixTO.builder()
+                .clusterPrefix(clusterPrefix)
+                .clusterRunningPrefix(clusterRunningPrefix)
+                .clusterWaitingPrefix(clusterWaitingPrefix)
+                .projectRunningKey(projectRunningKey)
+                .projectWaitingKey(projectWaitingKey)
+                .projectCheckKey(projectCheckKey)
+                .build();
+
+    }
+
+
+}

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/util/ScoreUtil.java

@@ -1,19 +0,0 @@
-package com.css.simulation.resource.scheduler.util;
-
-import com.css.simulation.resource.scheduler.pojo.to.ScoreTO;
-
-/**
- * 任务结果打分
- */
-public class ScoreUtil {
-
-
-
-
-    public static ScoreTO score(Object taskResult, String sceneType) {
-
-        return new ScoreTO();
-    }
-
-
-}