commit

simulation-resource-scheduler-agent/pom.xml

@@ -0,0 +1,164 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>simulation-cloud</artifactId>
+        <groupId>com.css</groupId>
+        <version>1.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>simulation-resource-scheduler-agent</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+
+
+        <!-- minio - 开始 -->
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.minio</groupId>
+            <artifactId>minio</artifactId>
+        </dependency>
+        <!-- minio - 结束 -->
+
+        <!-- 缓存 - 开始-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+        </dependency>
+        <!-- 缓存 - 结束-->
+
+        <!-- docker 客户端 -->
+        <dependency>
+            <groupId>com.github.docker-java</groupId>
+            <artifactId>docker-java</artifactId>
+        </dependency>
+
+        <!-- kubernetes 客户端 -->
+        <dependency>
+            <groupId>io.kubernetes</groupId>
+            <artifactId>client-java</artifactId>
+        </dependency>
+
+
+        <!-- apache kafka - 开始 -->
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <!-- apache kafka - 结束 -->
+
+        <!-- 数据库 - 开始 -->
+        <dependency>
+            <groupId>com.github.pagehelper</groupId>
+            <artifactId>pagehelper-spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.mybatis.spring.boot</groupId>
+            <artifactId>mybatis-spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>druid-spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+        </dependency>
+        <!-- 数据库 - 结束 -->
+
+        <!-- nacos - 开始 -->
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-bootstrap</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
+        </dependency>
+        <!-- nacos - 结束 -->
+
+        <!-- api-common -->
+        <dependency>
+            <groupId>com.css</groupId>
+            <artifactId>api-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <!-- 基础 -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-openfeign</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-loadbalancer</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-actuator</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>${spring-boot.version}</version>
+                <configuration>
+                    <fork>true</fork>
+                    <addResources>true</addResources>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

simulation-resource-scheduler-agent/src/main/java/com/css/simulation/resource/scheduler/agent/app/service/TaskAppService.java

@@ -2,7 +2,6 @@ package com.css.simulation.resource.scheduler.agent.app.service;
 
 import api.common.util.FileUtil;
 import api.common.util.OsUtil;
-import api.common.util.SchedulerUtil;
 import lombok.SneakyThrows;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;

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

@@ -2,11 +2,9 @@ package com.css.simulation.resource.scheduler.adapter.consumer;
 
 
 import api.common.util.JsonUtil;
-import com.css.simulation.resource.scheduler.adapter.controller.model.ProjectMessageModel;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStartMessageEntity;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStopMessageEntity;
 import com.css.simulation.resource.scheduler.app.service.ProjectApplicationService;
-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.springframework.kafka.annotation.KafkaListener;
@@ -14,34 +12,30 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 
-@Component
 @Slf4j
+@Component
 public class ProjectConsumer {
+
     @Resource
     private ProjectApplicationService projectApplicationService;
 
     /**
-     * @param projectRecord 项目启动消息
+     * @param projectStartMessageRecord 项目启动消息
      */
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${custom.mq-start-project-topic}")
-    @SneakyThrows
-    public void acceptMessage(ConsumerRecord<String, String> projectRecord) {
-        log.info("接收到项目开始消息为：" + projectRecord);
-        ProjectMessageModel projectMessageModel = JsonUtil.jsonToBean(projectRecord.value(), ProjectMessageModel.class);
-        projectApplicationService.runProject(projectMessageModel);
+    public void acceptMessage(ConsumerRecord<String, String> projectStartMessageRecord) {
+        log.info("消费者组 simulation-resource-scheduler 接收到项目开始消息：" + projectStartMessageRecord);
+        projectApplicationService.runProject(JsonUtil.jsonToBean(projectStartMessageRecord.value(), ProjectStartMessageEntity.class));
     }
 
-    //* -------------------------------- 结束 --------------------------------
 
+    /**
+     * @param projectStopMessageRecord 项目终止消息
+     */
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "${custom.mq-stop-project-topic}")
-    @SneakyThrows
-    public void stopProject(ConsumerRecord<String, String> stopRecord) {
-        log.info("接收到的项目终止消息为：" + stopRecord);
-        JsonNode jsonNode = new ObjectMapper().readTree(stopRecord.value());
-        String projectId = jsonNode.path("projectId").asText();
-        String projectType = jsonNode.path("type").asText();
-
-        projectApplicationService.stopProject(projectType, projectId);
+    public void stopProject(ConsumerRecord<String, String> projectStopMessageRecord) {
+        log.info("消费者组 simulation-resource-scheduler 接收到的项目终止消息：" + projectStopMessageRecord);
+        projectApplicationService.stopProject(JsonUtil.jsonToBean(projectStopMessageRecord.value(), ProjectStopMessageEntity.class));
     }
 
 

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/adapter/controller/model/ProjectMessageModel.java → simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/adapter/entity/ProjectStartMessageEntity.java

@@ -1,4 +1,4 @@
-package com.css.simulation.resource.scheduler.adapter.controller.model;
+package com.css.simulation.resource.scheduler.adapter.entity;
 
 import lombok.*;
 
@@ -18,7 +18,7 @@ import lombok.*;
 @NoArgsConstructor
 @AllArgsConstructor
 @EqualsAndHashCode
-public class ProjectMessageModel {
+public class ProjectStartMessageEntity {
 
     private String projectId;// 项目 id
     private String algorithmId;// 算法 id

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/adapter/entity/ProjectStopMessageEntity.java

@@ -0,0 +1,24 @@
+package com.css.simulation.resource.scheduler.adapter.entity;
+
+import lombok.*;
+
+/**
+ * {
+ * "projectId": "sadfasdfs", // 项目 id
+ * "algorithmId": "sadfasdfs", // 算法 id
+ * "vehicleId": "sadfasdfs", // 车辆 id
+ * "scenePackageId": "sadfasdfs", // 场景包 id
+ * "maxSimulationTime": 11111, // 最大仿真时间
+ * "parallelism": 30  // 并行度
+ * "type": 30  // 项目类型
+ * }
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode
+public class ProjectStopMessageEntity {
+    private String projectId;// 项目 id
+    private String type;// 项目类型
+}

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/app/entity/ProjectWaitQueueEntity.java

@@ -1,6 +1,6 @@
 package com.css.simulation.resource.scheduler.app.entity;
 
-import com.css.simulation.resource.scheduler.adapter.controller.model.ProjectMessageModel;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStartMessageEntity;
 import lombok.*;
 
 @EqualsAndHashCode
@@ -11,5 +11,5 @@ import lombok.*;
 public class ProjectWaitQueueEntity {
     private String waitingType; //1等待执行 2等待扩充
     private Integer waitingParallelism; // 等待扩充或执行的并行度
-    private ProjectMessageModel projectMessageModel; // 项目详情
+    private ProjectStartMessageEntity projectStartMessageEntity; // 项目详情
 }

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

@@ -2,7 +2,8 @@ package com.css.simulation.resource.scheduler.app.service;
 
 import api.common.pojo.constants.DictConstants;
 import api.common.util.*;
-import com.css.simulation.resource.scheduler.adapter.controller.model.ProjectMessageModel;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStartMessageEntity;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStopMessageEntity;
 import com.css.simulation.resource.scheduler.app.entity.ProjectWaitQueueEntity;
 import com.css.simulation.resource.scheduler.app.entity.VehicleEntity;
 import com.css.simulation.resource.scheduler.domain.service.ProjectDomainService;
@@ -103,33 +104,34 @@ public class ProjectApplicationService {
     // -------------------------------- Comment --------------------------------
 
     @Async("pool1")
-    public void runProject(ProjectMessageModel projectMessageModel) {
-        String projectId = projectMessageModel.getProjectId();
-        ProjectEntity project = projectDomainService.getProjectByProjectId(projectId);
-        if (project == null) {
-            throw new RuntimeException("不存在项目：" + projectId);
-        }
+    public void runProject(ProjectStartMessageEntity projectStartMessageEntity) {
         //1 创建任务文件并固定场景数据
-        createTaskAndFixData(projectMessageModel);
+        createTaskAndFixData(projectStartMessageEntity);
         //2 校验证书和并行度
-        checkIfCanRun(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_RUN_TYPE_EXECUTE).waitingParallelism(projectMessageModel.getParallelism()).projectMessageModel(projectMessageModel).build());
+        checkIfCanRun(
+                ProjectWaitQueueEntity.builder()
+                        .waitingType(DictConstants.PROJECT_RUN_TYPE_EXECUTE)
+                        .waitingParallelism(projectStartMessageEntity.getParallelism())
+                        .projectStartMessageEntity(projectStartMessageEntity)
+                        .build()
+        );
     }
 
     /**
      * 接收到运行信息立即复制一份数据作为据运行数
      *
-     * @param projectMessageModel 项目启动消息
+     * @param projectStartMessageEntity 项目启动消息
      */
-    public void createTaskAndFixData(ProjectMessageModel projectMessageModel) {
+    public void createTaskAndFixData(ProjectStartMessageEntity projectStartMessageEntity) {
         //* -------------------------------- 0 读取消息，创建临时目录 --------------------------------
-        String projectId = projectMessageModel.getProjectId();                // 手动执行项目 id 或 自动执行子项目 id
-        String projectType = projectMessageModel.getType();                   // 项目类型
+        String projectId = projectStartMessageEntity.getProjectId();                // 手动执行项目 id 或 自动执行子项目 id
+        String projectType = projectStartMessageEntity.getType();                   // 项目类型
         try {
-            String modelType = projectMessageModel.getModelType();                // 模型类型，1 动力学模型 2 carsim模型
-            String packageId = projectMessageModel.getScenePackageId();           // 场景测试包 id
-            String vehicleConfigId = projectMessageModel.getVehicleConfigId();    // 模型配置 id
-            String algorithmId = projectMessageModel.getAlgorithmId();            // 模型配置 id
-            long videoTime = projectMessageModel.getMaxSimulationTime();          // 结果视频的时长
+            String modelType = projectStartMessageEntity.getModelType();                // 模型类型，1 动力学模型 2 carsim模型
+            String packageId = projectStartMessageEntity.getScenePackageId();           // 场景测试包 id
+            String vehicleConfigId = projectStartMessageEntity.getVehicleConfigId();    // 模型配置 id
+            String algorithmId = projectStartMessageEntity.getAlgorithmId();            // 模型配置 id
+            long videoTime = projectStartMessageEntity.getMaxSimulationTime();          // 结果视频的时长
             String userId = projectDomainService.getUserIdByProjectIdAndProjectType(projectId, projectType);
             String projectPath = linuxTempPath + "project/" + projectId + "/";
             FileUtil.mkdir(projectPath);
@@ -142,8 +144,8 @@ public class ProjectApplicationService {
             //根据场景测试包 packageId，拿到场景集合（包括重复场景），重复场景会在发送消息时根据叶子指标发送多次。
             List<SceneEntity> sceneEntityList = getSceneList(projectId, packageId);
             int taskTotal = sceneEntityList.size();
-            projectMessageModel.setTaskTotal(taskTotal);
-            projectMessageModel.setTaskCompleted(0);
+            projectStartMessageEntity.setTaskTotal(taskTotal);
+            projectStartMessageEntity.setTaskCompleted(0);
             //去重，之后发送消息的时候会补全指标，如果不去重的话会出现多个场景重复关联多个指标
             Set<SceneEntity> sceneEntitySet = new HashSet<>(sceneEntityList);
             log.info("项目 " + projectId + " 场景包括：" + sceneEntitySet);
@@ -318,13 +320,13 @@ public class ProjectApplicationService {
     public void checkIfCanRun(ProjectWaitQueueEntity projectWaitQueueEntity) {
         final String waitingType = projectWaitQueueEntity.getWaitingType();
         final Integer waitingParallelism = projectWaitQueueEntity.getWaitingParallelism();
-        final ProjectMessageModel projectMessageModel = projectWaitQueueEntity.getProjectMessageModel();
-        log.debug("判断用户是否拥有可分配资源：" + projectMessageModel);
+        final ProjectStartMessageEntity projectStartMessageEntity = projectWaitQueueEntity.getProjectStartMessageEntity();
+        log.debug("判断用户是否拥有可分配资源：" + projectStartMessageEntity);
         //1 项目信息
-        String modelType = projectMessageModel.getModelType();
-        String projectId = projectMessageModel.getProjectId();    // 手动执行项目 id 或 自动执行子项目 id
-        int parallelism = projectMessageModel.getParallelism();   // 项目并行度
-        String projectType = projectMessageModel.getType(); // 项目类型
+        String modelType = projectStartMessageEntity.getModelType();
+        String projectId = projectStartMessageEntity.getProjectId();    // 手动执行项目 id 或 自动执行子项目 id
+        int parallelism = projectStartMessageEntity.getParallelism();   // 项目并行度
+        String projectType = projectStartMessageEntity.getType(); // 项目类型
         String isChoiceGpu = projectDomainService.getIsChoiceGpuByProjectId(projectId);
         //2 剩余并行度
         int remainderParallelism = projectDomainService.getRemainderParallelism(isChoiceGpu);
@@ -341,12 +343,12 @@ public class ProjectApplicationService {
             PrefixEntity redisPrefix = projectDomainService.getRedisPrefixByClusterIdAndProjectId(DictConstants.SYSTEM_CLUSTER_ID, projectId);
             final String projectRunningKey = redisPrefix.getProjectRunningKey();
             if (remainderParallelism <= 0) {
-                wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectMessageModel(projectMessageModel).build());
+                wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectStartMessageEntity(projectStartMessageEntity).build());
             } else if (remainderParallelism < parallelism) {
-                wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - remainderParallelism).projectMessageModel(projectMessageModel).build());
-                run(clusterUserId, remainderParallelism, projectMessageModel, projectRunningKey, isChoiceGpu);
+                wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - remainderParallelism).projectStartMessageEntity(projectStartMessageEntity).build());
+                run(clusterUserId, remainderParallelism, projectStartMessageEntity, projectRunningKey, isChoiceGpu);
             } else {
-                run(clusterUserId, parallelism, projectMessageModel, projectRunningKey, isChoiceGpu);
+                run(clusterUserId, parallelism, projectStartMessageEntity, projectRunningKey, isChoiceGpu);
             }
             return;
         } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
@@ -382,23 +384,23 @@ public class ProjectApplicationService {
         if (DictConstants.PROJECT_WAIT_TYPE_EXECUTE.equals(waitingType)) {
             if (DictConstants.MODEL_TYPE_VTD.equals(modelType)) {
                 if (remainderSimulationLicense <= 0 || remainderParallelism <= 0) {
-                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectMessageModel(projectMessageModel).build());
+                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectStartMessageEntity(projectStartMessageEntity).build());
                 } else if (remainderSimulationLicense < parallelism || remainderParallelism <= parallelism) {
-                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - Math.min(remainderSimulationLicense, remainderParallelism)).projectMessageModel(projectMessageModel).build());
-                    run(clusterUserId, Math.min(remainderSimulationLicense, remainderParallelism), projectMessageModel, projectRunningKey, isChoiceGpu);
+                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - Math.min(remainderSimulationLicense, remainderParallelism)).projectStartMessageEntity(projectStartMessageEntity).build());
+                    run(clusterUserId, Math.min(remainderSimulationLicense, remainderParallelism), projectStartMessageEntity, projectRunningKey, isChoiceGpu);
                 } else {
-                    run(clusterUserId, parallelism, projectMessageModel, projectRunningKey, isChoiceGpu);
+                    run(clusterUserId, parallelism, projectStartMessageEntity, projectRunningKey, isChoiceGpu);
                 }
             } else if (DictConstants.MODEL_TYPE_CARSIM.equals(modelType)) {
                 usingDynamicLicenseNumber = projectDomainService.getUsingLicenseNumber(clusterUserId, DictConstants.LICENSE_TYPE_DYNAMIC);
                 remainderDynamicLicense = numDynamicLicense - usingDynamicLicenseNumber;
                 if (remainderSimulationLicense <= 0 || remainderDynamicLicense <= 0 || remainderParallelism <= 0) {
-                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectMessageModel(projectMessageModel).build());
+                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXECUTE).waitingParallelism(parallelism).projectStartMessageEntity(projectStartMessageEntity).build());
                 } else if (remainderSimulationLicense < parallelism || remainderDynamicLicense < parallelism || remainderParallelism < parallelism) {
-                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - Math.min(Math.min(remainderSimulationLicense, remainderDynamicLicense), remainderParallelism)).projectMessageModel(projectMessageModel).build());
-                    run(clusterUserId, Math.min(remainderSimulationLicense, Math.min(remainderDynamicLicense, remainderParallelism)), projectMessageModel, projectRunningKey, isChoiceGpu);
+                    wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(parallelism - Math.min(Math.min(remainderSimulationLicense, remainderDynamicLicense), remainderParallelism)).projectStartMessageEntity(projectStartMessageEntity).build());
+                    run(clusterUserId, Math.min(remainderSimulationLicense, Math.min(remainderDynamicLicense, remainderParallelism)), projectStartMessageEntity, projectRunningKey, isChoiceGpu);
                 } else {
-                    run(clusterUserId, parallelism, projectMessageModel, projectRunningKey, isChoiceGpu);
+                    run(clusterUserId, parallelism, projectStartMessageEntity, projectRunningKey, isChoiceGpu);
                 }
             } else {
                 throw new RuntimeException("未知模型类型：" + modelType);
@@ -411,7 +413,7 @@ public class ProjectApplicationService {
                 } else {
                     if (remainderSimulationLicense < waitingParallelism || remainderParallelism <= waitingParallelism) {
                         expandParallelism = Math.min(remainderSimulationLicense, remainderParallelism);
-                        wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(waitingParallelism - expandParallelism).projectMessageModel(projectMessageModel).build());
+                        wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(waitingParallelism - expandParallelism).projectStartMessageEntity(projectStartMessageEntity).build());
                     } else {
                         expandParallelism = waitingParallelism;
                     }
@@ -426,7 +428,7 @@ public class ProjectApplicationService {
                 } else {
                     if (remainderSimulationLicense < waitingParallelism || remainderDynamicLicense < waitingParallelism || remainderParallelism < waitingParallelism) {
                         expandParallelism = Math.min(Math.min(remainderSimulationLicense, remainderDynamicLicense), remainderParallelism);
-                        wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(waitingParallelism - expandParallelism).projectMessageModel(projectMessageModel).build());
+                        wait(ProjectWaitQueueEntity.builder().waitingType(DictConstants.PROJECT_WAIT_TYPE_EXPAND).waitingParallelism(waitingParallelism - expandParallelism).projectStartMessageEntity(projectStartMessageEntity).build());
                     } else {
                         expandParallelism = waitingParallelism;
                     }
@@ -458,7 +460,7 @@ public class ProjectApplicationService {
             }
             boolean contains = false;
             for (ProjectWaitQueueEntity waitQueueEntity : waitingQueue) {
-                if (waitQueueEntity.getProjectMessageModel().getProjectId().equals(projectWaitQueueEntity.getProjectMessageModel().getProjectId())) {
+                if (waitQueueEntity.getProjectStartMessageEntity().getProjectId().equals(projectWaitQueueEntity.getProjectStartMessageEntity().getProjectId())) {
                     contains = true;
                     waitQueueEntity.setWaitingParallelism(projectWaitQueueEntity.getWaitingParallelism());
                 }
@@ -477,32 +479,32 @@ public class ProjectApplicationService {
     /**
      * 运行项目
      *
-     * @param projectMessageModel 初始接收到的项目启动信息
-     * @param projectRunningKey   projectRunningKey
+     * @param projectStartMessageEntity 初始接收到的项目启动信息
+     * @param projectRunningKey         projectRunningKey
      */
     @SneakyThrows
-    public void run(String clusterUserId, int finalParallelism, ProjectMessageModel projectMessageModel, String projectRunningKey, String isChoiceGpu) {
+    public void run(String clusterUserId, int finalParallelism, ProjectStartMessageEntity projectStartMessageEntity, String projectRunningKey, String isChoiceGpu) {
         Map<String, Integer> remainderNodeMap = projectDomainService.getRemainderNodeMap(isChoiceGpu);
-        String modelType = projectMessageModel.getModelType();
+        String modelType = projectStartMessageEntity.getModelType();
         if (!DictConstants.SYSTEM_USER_ID.equals(clusterUserId)) {
             projectDomainService.useLicense(clusterUserId, modelType, finalParallelism);
         }
-        String projectId = projectMessageModel.getProjectId();    // 项目 id
-        final String projectType = projectMessageModel.getType();
-        String vehicleConfigId = projectMessageModel.getVehicleConfigId();
+        String projectId = projectStartMessageEntity.getProjectId();    // 项目 id
+        final String projectType = projectStartMessageEntity.getType();
+        String vehicleConfigId = projectStartMessageEntity.getVehicleConfigId();
         String projectPath = linuxTempPath + "project/" + projectId + "/";
         // -------------------------------- 1 获取任务 json 列表 --------------------------------
         List<String> taskJsonList = FileUtil.listAbsolutePathByTypeAndLength(projectPath, "json", (StringUtil.getRandomUUID() + ".json").length());
         int taskTotal = taskJsonList.size();
-        projectMessageModel.setTaskTotal(taskTotal);
-        projectMessageModel.setTaskCompleted(0);
+        projectStartMessageEntity.setTaskTotal(taskTotal);
+        projectStartMessageEntity.setTaskCompleted(0);
         // 设置任务数量之后，获取运行节点，并将项目运行信息放入 redis
         //1 获取剩余并行度和即将使用的各node的并行度
         Map<String, Integer> nodeMap = projectDomainService.getNodeMapToUse(isChoiceGpu, finalParallelism);
         //2 将指定 node 的并行度减少
         nodeMap.keySet().forEach(nodeName -> projectDomainService.decrementParallelism(isChoiceGpu, nodeName, nodeMap.get(nodeName)));
         log.info("项目 " + projectId + " 运行在：" + nodeMap);
-        stringRedisTemplate.opsForValue().set(projectRunningKey, JsonUtil.beanToJson(projectMessageModel));
+        stringRedisTemplate.opsForValue().set(projectRunningKey, JsonUtil.beanToJson(projectStartMessageEntity));
         //* -------------------------------- 3 根据算法id查询算法名称 --------------------------------
 //        String algorithmDockerImage = projectDomainService.getAlgorithmDockerImageByProjectTypeAndProjectId(projectType, projectId);
         String algorithmDockerImage = projectDomainService.getAlgorithmDockerImageByProjectId(projectId);
@@ -774,11 +776,13 @@ public class ProjectApplicationService {
     }
 
     /**
-     * @param projectId   手动项目 id 或自动项目子id
-     * @param projectType 项目类型
+     * @param projectStopMessageEntity 项目终止消息实体
      */
     @SneakyThrows
-    public void stopProject(String projectType, String projectId) {
+    public void stopProject(ProjectStopMessageEntity projectStopMessageEntity) {
+        final String projectId = projectStopMessageEntity.getProjectId();
+        final String projectType = projectStopMessageEntity.getType();
+
         // 删除等待队列中的项目
         projectDomainService.removeWaitQueue(DictConstants.PROJECT_WAIT_TYPE_ALL, projectId);
         String isChoiceGpu = projectDomainService.getIsChoiceGpuByProjectId(projectId);

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

@@ -3,7 +3,7 @@ package com.css.simulation.resource.scheduler.domain.service;
 import api.common.pojo.common.ResponseBodyVO;
 import api.common.pojo.constants.DictConstants;
 import api.common.util.*;
-import com.css.simulation.resource.scheduler.adapter.controller.model.ProjectMessageModel;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStartMessageEntity;
 import com.css.simulation.resource.scheduler.app.entity.ProjectWaitQueueEntity;
 import com.css.simulation.resource.scheduler.infra.configuration.custom.CustomConfiguration;
 import com.css.simulation.resource.scheduler.infra.configuration.entity.NodeEntity;
@@ -724,15 +724,15 @@ public class ProjectDomainService {
     public boolean complete(PrefixEntity redisPrefix, String projectId) {
         boolean result = false;
         String projectMessageModelJson = customRedisClient.get(redisPrefix.getProjectRunningKey());
-        ProjectMessageModel projectMessageModel = JsonUtil.jsonToBean(projectMessageModelJson, ProjectMessageModel.class);
-        int taskTotal = projectMessageModel.getTaskTotal();
-        int taskCompleted = projectMessageModel.getTaskCompleted();
+        ProjectStartMessageEntity projectStartMessageEntity = JsonUtil.jsonToBean(projectMessageModelJson, ProjectStartMessageEntity.class);
+        int taskTotal = projectStartMessageEntity.getTaskTotal();
+        int taskCompleted = projectStartMessageEntity.getTaskCompleted();
         log.info("项目 " + projectId + " 完成进度为：" + (taskCompleted + 1) + "/" + taskTotal);
         if (taskCompleted + 1 == taskTotal) {
             result = true;
         } else {    // 项目没有完成
-            projectMessageModel.setTaskCompleted(taskCompleted + 1);  // 增加已完成任务数
-            stringRedisTemplate.opsForValue().set(redisPrefix.getProjectRunningKey(), JsonUtil.beanToJson(projectMessageModel));
+            projectStartMessageEntity.setTaskCompleted(taskCompleted + 1);  // 增加已完成任务数
+            stringRedisTemplate.opsForValue().set(redisPrefix.getProjectRunningKey(), JsonUtil.beanToJson(projectStartMessageEntity));
         }
         return result;
     }
@@ -937,7 +937,7 @@ public class ProjectDomainService {
         if (CollectionUtil.isNotEmpty(waitQueue)) {
             if (DictConstants.PROJECT_WAIT_TYPE_EXECUTE.equals(waitType)) {
                 for (ProjectWaitQueueEntity projectWaitQueueEntity : waitQueue) {
-                    if (projectWaitQueueEntity.getProjectMessageModel().getProjectId().equals(projectId) && projectWaitQueueEntity.getWaitingType().equals(DictConstants.PROJECT_WAIT_TYPE_EXECUTE)) {
+                    if (projectWaitQueueEntity.getProjectStartMessageEntity().getProjectId().equals(projectId) && projectWaitQueueEntity.getWaitingType().equals(DictConstants.PROJECT_WAIT_TYPE_EXECUTE)) {
                         waitQueue.remove(projectWaitQueueEntity);
                         break;
                     }
@@ -945,14 +945,14 @@ public class ProjectDomainService {
                 setWaitQueue(waitQueue);
             } else if (DictConstants.PROJECT_WAIT_TYPE_EXPAND.equals(waitType)) {
                 for (ProjectWaitQueueEntity projectWaitQueueEntity : waitQueue) {
-                    if (projectWaitQueueEntity.getProjectMessageModel().getProjectId().equals(projectId) && projectWaitQueueEntity.getWaitingType().equals(DictConstants.PROJECT_WAIT_TYPE_EXPAND)) {
+                    if (projectWaitQueueEntity.getProjectStartMessageEntity().getProjectId().equals(projectId) && projectWaitQueueEntity.getWaitingType().equals(DictConstants.PROJECT_WAIT_TYPE_EXPAND)) {
                         waitQueue.remove(projectWaitQueueEntity);
                         break;
                     }
                 }
             } else if (DictConstants.PROJECT_WAIT_TYPE_ALL.equals(waitType)) {
                 for (ProjectWaitQueueEntity projectWaitQueueEntity : waitQueue) {
-                    if (projectWaitQueueEntity.getProjectMessageModel().getProjectId().equals(projectId)) {
+                    if (projectWaitQueueEntity.getProjectStartMessageEntity().getProjectId().equals(projectId)) {
                         waitQueue.remove(projectWaitQueueEntity);
                         break;
                     }

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

@@ -2,7 +2,7 @@ package com.css.simulation.resource.scheduler.infra.configuration.scheduler;
 
 import api.common.pojo.constants.DictConstants;
 import api.common.util.CollectionUtil;
-import com.css.simulation.resource.scheduler.adapter.controller.model.ProjectMessageModel;
+import com.css.simulation.resource.scheduler.adapter.entity.ProjectStartMessageEntity;
 import com.css.simulation.resource.scheduler.app.entity.ProjectWaitQueueEntity;
 import com.css.simulation.resource.scheduler.app.service.ProjectApplicationService;
 import com.css.simulation.resource.scheduler.domain.service.ProjectDomainService;
@@ -36,8 +36,8 @@ public class ProjectScheduler {
                 projectApplicationService.checkIfCanRun(projectWaitQueueEntity);
             } else if (DictConstants.PROJECT_WAIT_TYPE_EXPAND.equals(waitingType)) {
                 log.debug("尝试扩充运行中的项目：{}", projectWaitQueueEntity);
-                ProjectMessageModel projectMessageModel = projectWaitQueueEntity.getProjectMessageModel();
-                String projectId = projectMessageModel.getProjectId();
+                ProjectStartMessageEntity projectStartMessageEntity = projectWaitQueueEntity.getProjectStartMessageEntity();
+                String projectId = projectStartMessageEntity.getProjectId();
                 ProjectEntity project = projectDomainService.getProjectByProjectId(projectId);
                 if (DictConstants.PROJECT_COMPLETED.equals(project.getNowRunState())) {
                     log.info("等待扩充的项目" + projectWaitQueueEntity + "已运行完成，无需扩充。");