mlx

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

@@ -26,8 +26,12 @@ public class ProjectMessageDTO {
     private String algorithmId;// 算法 id
     private String vehicleConfigId;// 车辆配置 id
     private String scenePackageId;// 场景包 id
-    private Long maxSimulationTime;// 最大仿真时间（秒）
-    private Long parallelism;// 并行度，创建 pod 时使用
+    private Integer maxSimulationTime;// 最大仿真时间（秒）
+    private Integer parallelism;// 期望并行度，页面上创建项目时指定的并行度
     private String type;// 项目类型
+    // -------------------------------- Comment --------------------------------
+    private Integer currentParallelism;// 当前正在使用的并行度
+    private Integer taskTotal;// 任务总数量
+    private Integer taskCompleted;// 任务已完成数量
 
 }

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

@@ -8,7 +8,6 @@ import api.common.util.JsonUtil;
 import api.common.util.StringUtil;
 import com.css.simulation.resource.scheduler.mapper.*;
 import com.css.simulation.resource.scheduler.pojo.po.*;
-import com.css.simulation.resource.scheduler.pojo.to.KubernetesNodeTO;
 import com.css.simulation.resource.scheduler.pojo.to.PrefixTO;
 import com.css.simulation.resource.scheduler.service.ProjectService;
 import com.css.simulation.resource.scheduler.util.ProjectUtil;
@@ -25,6 +24,7 @@ import org.springframework.stereotype.Component;
 import javax.annotation.Resource;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 @Component
@@ -33,10 +33,7 @@ public class ProjectConsumer {
 
     @Value("${scheduler.minio-path.project-result}")
     String resultPathMinio;
-    @Value("${scheduler.linux-path.job-template}")
-    String jobTemplate;
-    @Value("${scheduler.linux-path.job-yaml}")
-    String jobYaml;
+
 
     @Value("${scheduler.host.hostname}")
     String hostname;
@@ -77,10 +74,10 @@ public class ProjectConsumer {
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${scheduler.start-topic}")
     @SneakyThrows
     public void cacheProject(ConsumerRecord<String, String> projectRecord) {
-        String projectJson = projectRecord.value();
-        log.info("ProjectConsumer--cacheManualProject 接收到项目开始消息为：" + projectJson);
+        String initialProjectJson = projectRecord.value();
+        log.info("ProjectConsumer--cacheManualProject 接收到项目开始消息为：" + initialProjectJson);
         //1 读取 kafka 的 project 信息
-        ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
+        ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(initialProjectJson, ProjectMessageDTO.class);
         String projectId = projectMessageDTO.getProjectId();    // 手动执行项目 id 或 自动执行子项目 id
         long parallelism = projectMessageDTO.getParallelism();   // 项目并行度
         String projectType = projectMessageDTO.getType(); // 项目类型
@@ -91,11 +88,11 @@ public class ProjectConsumer {
         } else if (DictConstants.PROJECT_TYPE_AUTO_SUB.equals(projectType)) {
             userId = autoSubProjectMapper.selectCreateUserById(projectId);
         } else {
-            log.error("ProjectConsumer--cacheManualProject 项目类型错误：" + projectJson);
+            log.error("ProjectConsumer--cacheManualProject 项目类型错误：" + initialProjectJson);
             return;
         }
         if (StringUtil.isEmpty(userId)) {
-            log.error("ProjectConsumer--cacheManualProject 未查询到项目创建人：" + projectJson);
+            log.error("ProjectConsumer--cacheManualProject 未查询到项目创建人：" + initialProjectJson);
             return;
         }
         //3 获取用户类型（管理员账户、管理员子账户、普通账户、普通子账户）（独占、共享）
@@ -107,7 +104,7 @@ public class ProjectConsumer {
         if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
             log.info("ProjectConsumer--cacheManualProject 项目 " + projectId + " 的创建人 " + userId + " 为管理员账户或管理员子账户，直接判断服务器能否执行。");
             PrefixTO redisPrefix = projectUtil.getRedisPrefixByClusterIdAndProjectId(DictConstants.SYSTEM_CLUSTER_ID, projectId);
-            run(DictConstants.SYSTEM_CLUSTER_ID, projectId, projectType, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey(), projectJson, parallelism);
+            run(projectMessageDTO, DictConstants.SYSTEM_CLUSTER_ID, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             return;
         } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
             clusterPO = clusterMapper.selectByUserId(userId);
@@ -133,12 +130,12 @@ public class ProjectConsumer {
         Set<String> clusterRunningKeySet = stringRedisTemplate.keys(redisPrefix.getClusterRunningPrefix() + "*");
         List<String> runningProjectSet;
         if (CollectionUtil.isEmpty(clusterRunningKeySet)) {
-            run(clusterId, projectId, projectType, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey(), projectJson, parallelism);
+            run(projectMessageDTO, clusterId, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             return;
         }
         runningProjectSet = projectUtil.getRunningProjectList(clusterRunningKeySet);
         if (CollectionUtil.isEmpty(runningProjectSet)) {
-            run(clusterId, projectId, projectType, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey(), projectJson, parallelism);
+            run(projectMessageDTO, clusterId, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             return;
         }
         // 计算正在运行的项目的并行度总和
@@ -150,72 +147,67 @@ public class ProjectConsumer {
         }
         // 如果执行后的并行度总和小于最大节点数则执行，否则不执行
         if (parallelismSum + parallelism <= simulationLicenseNumber) {
-            run(clusterId, projectId, projectType, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey(), projectJson, parallelism);
+            run(projectMessageDTO, clusterId, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
         } else {
-            wait(clusterId, projectId, redisPrefix.getProjectWaitingKey(), projectJson);
+            wait(redisPrefix.getProjectWaitingKey(), projectMessageDTO);
         }
     }
 
     /**
-     * @param clusterId
-     * @param projectId
-     * @param projectType
-     * @param projectRunningKey
-     * @param projectJson
-     * @param parallelism
-     * @return
+     * @param projectMessageDTO 初始接收到的项目启动信息
+     * @param clusterId         集群 id
+     * @param projectRunningKey projectRunningKey
+     * @param projectWaitingKey projectWaitingKey
      */
-    public void run(String clusterId, String projectId, String projectType, String projectRunningKey, String projectWaitingKey, String projectJson, long parallelism) {
+    public void run(ProjectMessageDTO projectMessageDTO, String clusterId, String projectRunningKey, String projectWaitingKey) {
 
-        //1 获取一个剩余可用并行度最大的节点
-        KubernetesNodeTO maxParallelismNodeTO = projectUtil.getMaxParallelismNode();
-        String maxRestParallelismNode = maxParallelismNodeTO.getName();
-        long maxRestParallelism = maxParallelismNodeTO.getMaxParallelism();
-        log.info("ProjectConsumer--run 准备在节点 " + maxParallelismNodeTO + " 执行项目 " + projectId + "。");
+        String projectId = projectMessageDTO.getProjectId();
+        int parallelism = projectMessageDTO.getParallelism();  // 期望并行度
+        //1 获取所有节点的剩余可用并行度
+        Map<String, Integer> nodeMap = projectUtil.getNodeMap(parallelism);
+        if (nodeMap.size() == 0) {
+            return;
+        }
+        //2 计算实际可用并行度
+        int parallelismSum = nodeMap.keySet().stream().mapToInt(nodeMap::get).sum();
 
         //2 判断剩余可用并行度是否大于项目并行度，否则加入扩充队列
-        if (maxRestParallelism >= parallelism) {
-            log.info("ProjectConsumer--run 集群 " + clusterId + " 将项目 " + projectId + "在节点" + maxRestParallelismNode + " 执行！");
-            parseProject(projectId, projectType, projectJson, "cluster:" + clusterId, projectRunningKey, maxRestParallelismNode, parallelism);
-        } else if (maxRestParallelism > 0L) {
-            log.info("ProjectConsumer--run 集群 " + clusterId + " 将项目 " + projectId + "在节点" + maxRestParallelismNode + " 执行！");
-            parseProject(projectId, projectType, projectJson, "cluster:" + clusterId, projectRunningKey, maxRestParallelismNode, maxRestParallelism);
+        if (parallelismSum > 0L) {
+            log.info("ProjectConsumer--run 集群 " + clusterId + " 将项目 " + projectId + "在节点 " + nodeMap + " 上以并行度 " + parallelismSum + " 执行！");
+            projectMessageDTO.setCurrentParallelism(parallelismSum);    // 设置实际的并行度
+            parseProject(nodeMap, projectMessageDTO, "cluster:" + clusterId, projectRunningKey);
         } else {
-            wait(clusterId, projectId, projectWaitingKey, projectJson);
             log.info("ProjectConsumer--cacheManualProject 服务器资源不够，项目 " + projectId + " 暂时加入等待队列。");
+            wait(projectWaitingKey, projectMessageDTO);
         }
     }
 
     /**
-     * @param clusterId
-     * @param projectId
-     * @param projectWaitingKey
-     * @param projectJson
+     * @param projectWaitingKey 项目等待 key
+     * @param projectMessageDTO 项目信息
      */
-    public void wait(String clusterId, String projectId, String projectWaitingKey, String projectJson) {
-        log.info("ProjectConsumer--wait 集群 " + clusterId + " 将项目 " + projectId + " 放入等待队列！");
-        stringRedisTemplate.opsForValue().set(projectWaitingKey, projectJson);
+    @SneakyThrows
+    public void wait(String projectWaitingKey, ProjectMessageDTO projectMessageDTO) {
+        stringRedisTemplate.opsForValue().set(projectWaitingKey, JsonUtil.beanToJson(projectMessageDTO));
     }
 
 
     /**
-     * @param projectId
-     * @param projectJson
-     * @param clusterPrefix
-     * @param projectRunningPrefix projectRunningKey
-     * @param nodeName
-     * @param parallelism
+     * @param nodeMap           节点列表以及剩余可用并行度
+     * @param projectMessageDTO 初始接收到的项目启动信息
+     * @param clusterPrefix     clusterPrefix
+     * @param projectRunningKey projectRunningKey
      */
     @SneakyThrows
-    public void parseProject(String projectId, String projectType, String projectJson, String clusterPrefix, String projectRunningPrefix, String nodeName, long parallelism) {
-        // -------------------------------- 0 准备 --------------------------------
-        projectService.prepare(clusterPrefix, projectId, projectType, projectRunningPrefix, projectJson, nodeName, parallelism);
-        log.info("ProjectConsumer--parseManualProject 接收到项目开始消息为：" + projectJson);
-        ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
+    public void parseProject(Map<String, Integer> nodeMap, ProjectMessageDTO projectMessageDTO, String clusterPrefix, String projectRunningKey) {
+        String projectId = projectMessageDTO.getProjectId();    // 项目 id
+        String projectType = projectMessageDTO.getType();   // 项目类型
         String packageId = projectMessageDTO.getScenePackageId();   // 场景测试包 id
-        Long maxSimulationTime = projectMessageDTO.getMaxSimulationTime(); // 最大仿真时间，即生成视频的时间长度
+        long videoTime = projectMessageDTO.getMaxSimulationTime(); // 结果视频的时长
         String vehicleConfigId = projectMessageDTO.getVehicleConfigId();// 模型配置 id
         String algorithmId = projectMessageDTO.getAlgorithmId();    // 算法 id
+        // -------------------------------- 0 准备 --------------------------------
+        projectService.prepare(nodeMap, projectMessageDTO, clusterPrefix, projectRunningKey);
         String userId = null;
         if (DictConstants.PROJECT_TYPE_MANUAL.equals(projectType)) {
             userId = manualProjectMapper.selectCreateUserById(projectId);
@@ -224,27 +216,26 @@ public class ProjectConsumer {
         }
         // -------------------------------- 1 查询场景 --------------------------------
         //1-1 根据场景测试包 packageId，拿到场景集合（不包括重复场景），重复场景会在发送消息时根据叶子指标发送多次。
-        List<ScenePO> scenePOList = projectService.handlePackage(projectRunningPrefix, projectId, packageId);
+        List<ScenePO> scenePOList = projectService.handlePackage(projectRunningKey, projectId, packageId);
+        int taskTotal = scenePOList.size();
+        projectMessageDTO.setTaskTotal(taskTotal);
+        projectMessageDTO.setTaskCompleted(0);
+        // 设置任务数量之后将项目运行信息放入 redis
+        stringRedisTemplate.opsForValue().set(projectRunningKey, JsonUtil.beanToJson(projectMessageDTO));
         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 发送任务消息 --------------------------------
-        projectService.sendTaskMessage(projectRunningPrefix, userId, projectId, projectType, maxSimulationTime, scenePOSet, vehiclePO, cameraPOList, ogtPOList);
+        projectService.sendTaskMessage(projectRunningKey, userId, projectId, projectType, videoTime, scenePOSet, vehiclePO, cameraPOList, ogtPOList);
         // -------------------------------- 4 算法导入（一期按单机版做） --------------------------------
         String algorithmDockerImage = projectService.handleAlgorithm(projectId, algorithmId);
         // -------------------------------- 5 创建 pod 开始执行 --------------------------------
-        projectService.transferAndRunYaml(
-                nodeName,
-                jobTemplate + "job-template.yaml",
-                projectId,
-                algorithmDockerImage,
-                scenePOList.size(),
-                parallelism,
-                jobYaml + "project-" + projectId + ".yaml"
-        );
+        //TODO
+        projectService.transferAndRunYaml(projectId, nodeMap, algorithmDockerImage);
 
     }
 

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

@@ -236,7 +236,7 @@ public class TaskManager {
                         try {
                             log.info("TaskService--state 开始执行打分命令：" + scoreCommand);
                             scoreResult = SshUtil.execute(session, scoreCommand);
-                            log.info("TaskService--state 项目" + projectId + "的任务" + task2Id + "打分结束，结果为：" + scoreResult);
+                            log.info("TaskService--state 项目" + projectId + " 的任务 " + task2Id + " 打分结束，结果为：" + scoreResult);
                             String replace = StringUtil.replace(scoreResult, "'", "\"");
                             score = JsonUtil.jsonToBean(replace, ScoreTO.class);
                             FileUtil.rm(runResultLinux);

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

@@ -11,8 +11,10 @@ import lombok.NoArgsConstructor;
 @AllArgsConstructor
 public class KubernetesNodeTO {
     private String name;
-    private Long maxParallelism;
+    private Integer maxParallelism;
     private String hostname;
     private String username;
     private String password;
+
+
 }

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

@@ -1,8 +1,8 @@
 package com.css.simulation.resource.scheduler.service;
 
 import api.common.pojo.constants.DictConstants;
+import api.common.pojo.dto.ProjectMessageDTO;
 import api.common.util.*;
-import com.css.simulation.resource.scheduler.configuration.kubernetes.KubernetesConfiguration;
 import com.css.simulation.resource.scheduler.mapper.*;
 import com.css.simulation.resource.scheduler.pojo.po.*;
 import com.css.simulation.resource.scheduler.pojo.to.*;
@@ -36,6 +36,10 @@ public class ProjectService {
     String projectResultPathOfMinio;
     @Value("${scheduler.linux-path.temp}")
     String linuxTempPath;
+    @Value("${scheduler.linux-path.pod-template-yaml}")
+    String podTemplateYaml;
+    @Value("${scheduler.linux-path.pod-yaml-directory}")
+    String podYamlDirectory;
     @Value("${minio.bucket-name}")
     String bucketName;
     @Value("${scheduler.host.hostname}")
@@ -68,21 +72,26 @@ public class ProjectService {
     SceneMapper sceneMapper;
     @Resource
     AlgorithmMapper algorithmMapper;
-    @Resource
-    KubernetesConfiguration kubernetesConfiguration;
+
 
     // -------------------------------- Comment --------------------------------
 
+    /**
+     * @param nodeMap           节点列表以及剩余可用并行度
+     * @param projectMessageDTO 初始接收到的项目启动信息
+     * @param clusterPrefix     clusterPrefix
+     * @param projectRunningKey projectRunningKey
+     */
     @Transactional
-    public void prepare(String clusterPrefix, String projectId, String projectType, String projectRunningKey, String projectJson, String nodeName, long parallelism) {
+    public void prepare(Map<String, Integer> nodeMap, ProjectMessageDTO projectMessageDTO, String clusterPrefix, String projectRunningKey) {
+        String projectId = projectMessageDTO.getProjectId();
         //1 将指定 node 的并行度减少
-        String restParallelismKey = "node:" + nodeName + ":parallelism";
-        String usedParallelismKey = "project:" + projectId + ":parallelism";
-        String nodeOfProject = "project:" + projectId + ":node";
-        long restParallelism = Long.parseLong(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(restParallelismKey)));// 剩余可用并行度
-        stringRedisTemplate.opsForValue().set(restParallelismKey, (restParallelism - parallelism) + "");
-        stringRedisTemplate.opsForValue().set(usedParallelismKey, parallelism + "");
-        stringRedisTemplate.opsForValue().set(nodeOfProject, nodeName);
+        nodeMap.keySet().forEach(nodeName -> {
+            long parallelismToUse = nodeMap.get(nodeName);
+            String restParallelismKey = "node:" + nodeName + ":parallelism";
+            long restParallelism = Long.parseLong(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(restParallelismKey)));// 剩余可用并行度
+            stringRedisTemplate.opsForValue().set(restParallelismKey, (restParallelism - parallelismToUse) + "");
+        });
         //2 将 redis 中该项目旧的信息则直接删除（包括 waitingKey)
         Set<String> oldKeys = stringRedisTemplate.keys(clusterPrefix + "*");
         if (CollectionUtil.isNotEmpty(oldKeys)) {
@@ -92,8 +101,7 @@ public class ProjectService {
                 }
             }
         }
-        //3 将项目信息放入 redis
-        stringRedisTemplate.opsForValue().set(projectRunningKey, projectJson);
+
         //4 将项目状态修改为执行中
         if (DictConstants.PROJECT_TYPE_MANUAL.equals(projectType)) {
             manualProjectMapper.updateNowRunStateById(DictConstants.PROJECT_RUNNING, projectId);   // 修改该 project 的状态为执行中，同时将已完成任务重置为 0 。
@@ -109,9 +117,9 @@ public class ProjectService {
 
     @SneakyThrows
     @Transactional
-    public List<ScenePO> handlePackage(String projectRunningPrefix, String projectId, String packageId) {
-        String allIndexPrefix = projectRunningPrefix + ":package:" + packageId + ":all";
-        String leafIndexPrefix = projectRunningPrefix + ":package:" + packageId + ":leaf";
+    public List<ScenePO> handlePackage(String projectRunningKey, String projectId, String packageId) {
+        String allIndexPrefix = projectRunningKey + ":package:" + packageId + ":all";
+        String leafIndexPrefix = projectRunningKey + ":package:" + packageId + ":leaf";
 
         //1 查询该场景包的所有指标列表存入 redis，包删了无所谓，但要过滤删掉的指标
         List<IndexTemplatePO> allIndexList = indexTemplateMapper.selectByPackageIdIncludeDeleted(packageId);
@@ -159,7 +167,7 @@ public class ProjectService {
      * @param cameraPOList
      * @param ogtPOList
      */
-    public void sendTaskMessage(String projectRunningPrefix, String userId, String projectId, String projectType, Long maxSimulationTime, Set<ScenePO> scenePOSet, VehiclePO vehiclePO, List<CameraPO> cameraPOList, List<OgtPO> ogtPOList) {
+    public void sendTaskMessage(String projectRunningPrefix, String userId, String projectId, String projectType, Long videoTime, Set<ScenePO> scenePOSet, VehiclePO vehiclePO, List<CameraPO> cameraPOList, List<OgtPO> ogtPOList) {
         final int[] messageNumber = {0};
         log.info("ProjectService--sendTaskMessage 项目 " + projectId + " 获得的包括的场景信息为：" + scenePOSet);
         for (ScenePO scenePO : scenePOSet) {
@@ -202,7 +210,7 @@ public class ProjectService {
                                 .project_id(taskPO.getPId())
                                 .task_id(taskPO.getId())
                                 .task_path(taskPO.getRunResultFilePath())
-                                .default_time(maxSimulationTime)
+                                .default_time(videoTime)
                                 .build())
                         .scenario(ScenarioTO.builder()
                                 .scenario_osc(scenePO.getScenarioOsc())
@@ -271,6 +279,7 @@ public class ProjectService {
 
     /**
      * 将 master 节点设置成镜像仓库，导入镜像的同时 commit 到仓库当中，供其他节点 pull
+     *
      * @param projectId
      * @param algorithmId
      * @return
@@ -368,18 +377,13 @@ public class ProjectService {
     /**
      * 运行
      *
-     * @param jobTemplateYamlPathSource 模板文件
-     * @param projectId                 项目id
-     * @param algorithmDockerImage      算法镜像
-     * @param completions               完成度
-     * @param parallelism               并行度
-     * @param jobTemplateYamlPathTarget 执行文件
+     * @param projectId            项目id
+     * @param nodeMap              并行度
+     * @param algorithmDockerImage 算法镜像
      */
     @SneakyThrows
-    public void transferAndRunYaml(String nodeName, String jobTemplateYamlPathSource, String projectId, String algorithmDockerImage, long completions, long parallelism, String jobTemplateYamlPathTarget) {
-        log.info("ProjectConsumer--transferYaml 项目 " + projectId + " 的完成度为：" + completions);
-        log.info("ProjectConsumer--transferYaml 项目 " + projectId + " 的并行度为：" + parallelism);
-        String yamlSource = FileUtil.read(jobTemplateYamlPathSource);
+    public void transferAndRunYaml(String projectId, Map<String, Integer> nodeMap, String algorithmDockerImage) {
+        String yamlSource = FileUtil.read(podTemplateYaml);
 //        log.info("ProjectConsumer--transferYaml 模板文件为：" + yamlSource);
         String replace0 = yamlSource.replace("job-cloud-simulation", "project-" + projectId);
         String replace1 = replace0.replace("vtd-container", "vtd-" + projectId);
@@ -391,7 +395,7 @@ public class ProjectService {
         String replace7 = replace6.replace("apiVers1on", "apiVersion");
         String replace8 = replace7.replace("1atch/v1", "batch/v1");
         String finalYaml = replace8.replace("node-name", nodeName);
-        log.info("ProjectConsumer--parseManualProject 在节点 " + nodeName + " 开始执行 job：" + finalYaml);
+        log.info("ProjectConsumer--parseManualProject 在节点 " + nodeName + " 开始执行 pod：" + finalYaml);
         FileUtil.writeStringToLocalFile(finalYaml, jobTemplateYamlPathTarget);
         //  启动
         KubernetesUtil.applyYaml(hostname, username, password, jobTemplateYamlPathTarget);

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

@@ -16,8 +16,7 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -43,32 +42,47 @@ public class ProjectUtil {
     StringRedisTemplate stringRedisTemplate;
 
 
-    public KubernetesNodeTO getMaxParallelismNode() {
-        List<KubernetesNodeTO> nodeList = kubernetesConfiguration.getNodeList();
-//        log.info("ProjectUtil--getMaxParallelismNode kubernetes 节点列表为：" + nodeList);
-        String maxRestParallelismNode = "master";
-        long maxRestParallelism = 0L;
-        for (KubernetesNodeTO kubernetesNodeTO : nodeList) {
+    /**
+     * 根据并行度获取用于执行的节点列表
+     * 根据剩余可用并行度降序排序
+     *
+     * @return 节点映射（节点名，并行度）
+     */
+    public Map<String, Integer> getNodeMap(long parallelism) {
+        List<KubernetesNodeTO> initialNodeList = kubernetesConfiguration.getNodeList(); // 预设并行度的节点列表
+        List<KubernetesNodeTO> restNodeList = new ArrayList<>();    // 剩余并行度的节点列表
+        Map<String, Integer> resultNodeMap = new HashMap<>();    // 用于执行的节点映射（节点名，并行度）
+        for (KubernetesNodeTO kubernetesNodeTO : initialNodeList) {
             String name = kubernetesNodeTO.getName();
-            long maxParallelism = kubernetesNodeTO.getMaxParallelism();
+            int maxParallelism = kubernetesNodeTO.getMaxParallelism();
             String restParallelismKey = "node:" + name + ":parallelism";
             String restParallelismString = stringRedisTemplate.opsForValue().get(restParallelismKey);
-            long restParallelism;
-            if (restParallelismString == null) {    // 如果剩余可用并行度没有值，说明是第一次查询，则重置成最大剩余可用并行度
+            int restParallelism;
+            if (restParallelismString == null) {    // 如果剩余可用并行度没有值，说明是第一次查询，则重置成最大并行度的预设值
                 restParallelism = maxParallelism;
                 stringRedisTemplate.opsForValue().set(restParallelismKey, restParallelism + "");
             } else {
-                restParallelism = Long.parseLong(restParallelismString);
+                restParallelism = Integer.parseInt(restParallelismString);
+                kubernetesNodeTO.setMaxParallelism(restParallelism);
             }
-            if (restParallelism >= maxRestParallelism) {
-                maxRestParallelism = restParallelism;
-                maxRestParallelismNode = name;
+            if (restParallelism > 0) {
+                restNodeList.add(kubernetesNodeTO);
+                resultNodeMap.put(name, 0);
             }
         }
-        return KubernetesNodeTO.builder()
-                .name(maxRestParallelismNode)
-                .maxParallelism(maxRestParallelism)
-                .build();
+
+        for (int i = 0; i < parallelism; i++) {
+            // 每次降序排序都取剩余并行度最大的一个。
+            restNodeList.sort((o1, o2) -> (int) (o2.getMaxParallelism() - o1.getMaxParallelism()));
+            KubernetesNodeTO tempNode = restNodeList.get(0);
+            String tempNodeName = tempNode.getName();
+            int tempParallelism = tempNode.getMaxParallelism();
+            if (tempParallelism > 0) {
+                tempNode.setMaxParallelism(tempParallelism - 1);
+                resultNodeMap.put(tempNodeName, resultNodeMap.get(tempNodeName) + 1);
+            }
+        }
+        return resultNodeMap;
     }
 
 

simulation-resource-server/src/main/java/com/css/simulation/resource/project/impl/SimulationProjectServiceImpl.java

@@ -1893,7 +1893,7 @@ public class SimulationProjectServiceImpl implements SimulationProjectService {
         }
 
         Map<String, String> map1 = new HashMap<>();
-        map1.put("label", "场景ID");
+        map1.put("label", "场景名称");
         map1.put("prop", "sceneId");
         cloums.add(map1);