调度BUG

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

@@ -2,6 +2,8 @@ package api.common.pojo.constants;
 
 public class DictConstants {
 
+    public static final String LICENSE_TYPE_SIMULATION = "1";
+    public static final String LICENSE_TYPE_DYNAMIC = "2";
     public static final String MODEL_TYPE_VTD = "1";
     public static final String MODEL_TYPE_CARSIM = "2";
     public static final String SCHEDULER_USER_ID = "simulation-resource-scheduler";

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

@@ -300,7 +300,7 @@ public class ProjectConsumer {
      */
     @SneakyThrows
     public void cacheProject(ProjectMessageDTO projectMessageDTO) {
-        log.info("判断用户是否拥有可分配资源：" + projectMessageDTO);
+        log.debug("判断用户是否拥有可分配资源：" + projectMessageDTO);
         //1 读取 kafka 的 project 信息
         final String modelType = projectMessageDTO.getModelType();
         String projectId = projectMessageDTO.getProjectId();    // 手动执行项目 id 或 自动执行子项目 id
@@ -316,23 +316,23 @@ public class ProjectConsumer {
         String clusterUserId;  // 项目实际运行使用的用户集群
         if (DictConstants.ROLE_CODE_SYSADMIN.equals(roleCode) || DictConstants.ROLE_CODE_ADMIN.equals(roleCode)) {  //3-1 管理员账户和管理员子账户直接执行
             clusterUserId = DictConstants.SYSTEM_CLUSTER_ID;
-            log.info("项目 " + projectId + " 的创建人 " + projectUserId + " 为管理员账户或管理员子账户，直接判断服务器能否执行。");
+            log.debug("项目 " + projectId + " 的创建人 " + projectUserId + " 为管理员账户或管理员子账户，直接判断服务器能否执行。");
             PrefixEntity redisPrefix = projectUtil.getRedisPrefixByClusterIdAndProjectId(DictConstants.SYSTEM_CLUSTER_ID, projectId);
             run(projectMessageDTO, clusterUserId, modelType, DictConstants.SYSTEM_USER_ID, redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             return;
         } else if (DictConstants.ROLE_CODE_UESR.equals(roleCode)) { //3-2 普通账户，不管是独占还是共享，都在自己的集群里排队，根据自己的独占节点排队
             clusterUserId = projectUserId;
             clusterEntity = clusterMapper.selectByUserId(clusterUserId);
-            log.info("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通账户（包括独占或共享都在自己的集群），集群为：" + clusterEntity);
+            log.debug("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通账户（包括独占或共享都在自己的集群），集群为：" + clusterEntity);
         } else if (DictConstants.ROLE_CODE_SUBUESR.equals(roleCode)) {
             if (DictConstants.USER_TYPE_EXCLUSIVE.equals(useType)) {   //3-3 普通子账户，根据自己的独占节点排队
                 clusterUserId = projectUserId;
                 clusterEntity = clusterMapper.selectByUserId(clusterUserId);
-                log.info("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通独占子账户（自己的集群），集群为：" + clusterEntity);
+                log.debug("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通独占子账户（自己的集群），集群为：" + clusterEntity);
             } else if (DictConstants.USER_TYPE_PUBLIC.equals(useType)) {    //3-4 共享子账户，根据父账户的共享节点排队
                 clusterUserId = userEntity.getCreateUserId();
                 clusterEntity = clusterMapper.selectByUserId(clusterUserId);
-                log.info("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通共享子账户（父账户的集群），集群为：" + clusterEntity);
+                log.debug("项目 " + projectId + " 的创建人 " + projectUserId + " 为普通共享子账户（父账户的集群），集群为：" + clusterEntity);
             } else {
                 throw new RuntimeException("用户" + projectUserId + "未知占用类型：" + useType);
             }
@@ -340,15 +340,20 @@ public class ProjectConsumer {
             throw new RuntimeException("未知角色类型：" + roleCode);
         }
         PrefixEntity redisPrefix = projectUtil.getRedisPrefixByClusterIdAndProjectId(clusterEntity.getId(), projectId);
+        final Integer usingSimulationLicenseNumber = projectUtil.getUsingLicenseNumber(clusterUserId, DictConstants.LICENSE_TYPE_SIMULATION);
+        final Integer usingDynamicLicenseNumber;
+        final Integer numSimulationLicense = clusterEntity.getNumSimulationLicense();
+        final Integer numDynamicLicense = clusterEntity.getNumDynamicLicense();
+        final String clusterId = clusterEntity.getId();
         if (DictConstants.MODEL_TYPE_VTD.equals(modelType)) {
             // 获取仿真软件证书数量和动力学软件证书数量（vtd占一个仿真证书，carsim各占一个）
 //            // 获取正在运行的项目的并行度总和
 //            int currentParallelismSum = projectUtil.getCurrentParallelismSum(redisPrefix.getClusterRunningPrefix());
             // 如果执行后的并行度总和小于最大节点数则执行，否则不执行
-            if (projectUtil.getUsingLicenseNumber(clusterUserId, DictConstants.MODEL_TYPE_VTD) + parallelism <= clusterEntity.getNumSimulationLicense()) {
+            if (usingSimulationLicenseNumber + parallelism <= numSimulationLicense) {
                 run(projectMessageDTO, clusterUserId, modelType, clusterEntity.getId(), redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             } else {
-                log.info("VTD 项目 " + projectId + " 并行度超出账户允许，加入等待队列，暂不执行。 ");
+                log.info("VTD 项目 {} 并行度为 {}，用户 {} 的集群 {} 的仿真证书总数量为 {}，已使用数量为 {}。该项目加入等待队列。 ", projectId, parallelism, clusterUserId, clusterId, numSimulationLicense, usingSimulationLicenseNumber);
                 wait(redisPrefix.getProjectWaitingKey(), projectMessageDTO);
             }
         } else if (DictConstants.MODEL_TYPE_CARSIM.equals(modelType)) {
@@ -356,10 +361,11 @@ public class ProjectConsumer {
 //                // 获取正在运行的项目的并行度总和
 //                int currentParallelismSum = projectUtil.getCurrentParallelismSum(redisPrefix.getClusterRunningPrefix());
             // 如果执行后的并行度总和小于最大节点数则执行，否则不执行
-            if (projectUtil.getUsingLicenseNumber(clusterUserId, DictConstants.MODEL_TYPE_VTD) + parallelism <= clusterEntity.getNumSimulationLicense() && projectUtil.getUsingLicenseNumber(clusterUserId, DictConstants.MODEL_TYPE_CARSIM) + parallelism <= clusterEntity.getNumDynamicLicense()) {
+            usingDynamicLicenseNumber = projectUtil.getUsingLicenseNumber(clusterUserId, DictConstants.LICENSE_TYPE_DYNAMIC);
+            if (usingSimulationLicenseNumber + parallelism <= numSimulationLicense && usingDynamicLicenseNumber + parallelism <= numDynamicLicense) {
                 run(projectMessageDTO, clusterUserId, modelType, clusterEntity.getId(), redisPrefix.getProjectRunningKey(), redisPrefix.getProjectWaitingKey());
             } else {
-                log.info("CARSIM 项目 " + projectId + " 并行度超出账户允许，加入等待队列，暂不执行。 ");
+                log.info("CARSIM 项目 {} 并行度为 {}，用户 {} 的集群 {} 的仿真证书总数量为 {}，已使用数量为 {}，动力学证书总数量为 {}，已使用数量为 {}。该项目加入等待队列。 ", projectId, parallelism, clusterUserId, clusterId, numSimulationLicense, usingSimulationLicenseNumber, numDynamicLicense, usingDynamicLicenseNumber);
                 wait(redisPrefix.getProjectWaitingKey(), projectMessageDTO);
             }
         } else {

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

@@ -1,12 +1,18 @@
 package com.css.simulation.resource.scheduler.scheduler;
 
+import api.common.pojo.dto.ProjectMessageDTO;
+import api.common.util.CollectionUtil;
+import api.common.util.JsonUtil;
 import com.css.simulation.resource.scheduler.consumer.ProjectConsumer;
 import com.css.simulation.resource.scheduler.util.ProjectUtil;
+import com.css.simulation.resource.scheduler.util.RedisUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
+import java.util.List;
 
 @Component
 @Slf4j
@@ -18,20 +24,19 @@ public class ProjectScheduler {
     @Resource
     private ProjectUtil projectUtil;
 
-//    /**
-//     * 调度项目启动
-//     */
-//    @Scheduled(fixedDelay = 1000)
-//    public void dispatchProject() {
-//        List<String> projectMessageKeys = projectUtil.getWaitingProjectMessageKeys();
-//        if (!CollectionUtil.isEmpty(projectMessageKeys)) {
-//            log.info("尝试启动等待中的项目：" + projectMessageKeys);
-//            for (String projectMessageKey : projectMessageKeys) {
-//                final String projectMessage = RedisUtil.getStringByKey(stringRedisTemplate, projectMessageKey);
-//                ConsumerRecord<String, String> projectRecord = new ConsumerRecord<>("", 0, 0L, null, projectMessage);
-//                projectConsumer.acceptMessage(projectRecord);
-//            }
-//        }
-//    }
+    /**
+     * 调度项目启动
+     */
+    @Scheduled(fixedDelay = 1000)
+    public void dispatchProject() {
+        List<String> projectMessageKeys = projectUtil.getWaitingProjectMessageKeys();
+        if (!CollectionUtil.isEmpty(projectMessageKeys)) {
+            log.info("尝试启动等待中的项目：" + projectMessageKeys);
+            for (String projectMessageKey : projectMessageKeys) {
+                final String projectMessage = RedisUtil.getStringByKey(stringRedisTemplate, projectMessageKey);
+                projectConsumer.cacheProject(JsonUtil.jsonToBean(projectMessage, ProjectMessageDTO.class));
+            }
+        }
+    }
 
 }

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

@@ -487,7 +487,6 @@ public class ProjectUtil {
     }
 
 
-
     public PrefixEntity getRedisPrefixByProjectIdAndProjectType(String projectId, String projectType) {
         String userId;
         if (DictConstants.PROJECT_TYPE_MANUAL.equals(projectType)) {
@@ -618,14 +617,14 @@ public class ProjectUtil {
     }
 
 
-    public Integer getUsingLicenseNumber(String userId, String modelType) {
+    public Integer getUsingLicenseNumber(String userId, String licenseType) {
         String key;
-        if (DictConstants.MODEL_TYPE_VTD.equals(modelType)) {
-            key = "user:" + userId + ":using-license:vtd";
-        } else if (DictConstants.MODEL_TYPE_CARSIM.equals(modelType)) {
-            key = "user:" + userId + ":using-license:carsim";
+        if (DictConstants.LICENSE_TYPE_SIMULATION.equals(licenseType)) {
+            key = "user:" + userId + ":using-license:simulation";
+        } else if (DictConstants.LICENSE_TYPE_DYNAMIC.equals(licenseType)) {
+            key = "user:" + userId + ":using-license:dynamic";
         } else {
-            throw new RuntimeException("未知模型类型：" + modelType);
+            throw new RuntimeException("未知证书类型：" + licenseType);
         }
         final String usingLicense = customRedisClient.get(key);
         if (StringUtil.isEmpty(usingLicense)) {
@@ -636,20 +635,23 @@ public class ProjectUtil {
         }
     }
 
-    public void useLicense(String userId, String modelType, int parallelism) {
-        String key;
+    public void useLicense(String clusterUserId, String modelType, int parallelism) {
         if (DictConstants.MODEL_TYPE_VTD.equals(modelType)) {
-            key = "user:" + userId + ":using-license:vtd";
+            useLicense("user:" + clusterUserId + ":using-license:simulation", parallelism);
         } else if (DictConstants.MODEL_TYPE_CARSIM.equals(modelType)) {
-            key = "user:" + userId + ":using-license:carsim";
+            useLicense("user:" + clusterUserId + ":using-license:simulation", parallelism);
+            useLicense("user:" + clusterUserId + ":using-license:dynamic", parallelism);
         } else {
             throw new RuntimeException("未知模型类型：" + modelType);
         }
+    }
+
+    private void useLicense(String key, int number) {
         final String usingLicense = customRedisClient.get(key);
         if (StringUtil.isEmpty(usingLicense)) {
-            customRedisClient.set(key, String.valueOf(parallelism));
+            customRedisClient.set(key, String.valueOf(number));
         } else {
-            customRedisClient.increment(key, parallelism);
+            customRedisClient.increment(key, number);
         }
     }
 
@@ -657,33 +659,24 @@ public class ProjectUtil {
     public void releaseLicense(String projectUserId, String modelType, int parallelism) {
         final String clusterUserId = getClusterUserIdByProjectUserId(projectUserId);
         if (DictConstants.MODEL_TYPE_VTD.equals(modelType)) {
-            String key = "user:" + clusterUserId + ":using-license:vtd";
-            final String usingLicense = customRedisClient.get(key);
-            if (StringUtil.isEmpty(usingLicense)) {
-                customRedisClient.set(key, "0");
-            } else {
-                customRedisClient.decrement(key, parallelism);
-            }
+            releaseLicense("user:" + clusterUserId + ":using-license:simulation",parallelism);
         } else if (DictConstants.MODEL_TYPE_CARSIM.equals(modelType)) {
-            String key1 = "user:" + clusterUserId + ":using-license:vtd";
-            final String usingLicense1 = customRedisClient.get(key1);
-            if (StringUtil.isEmpty(usingLicense1)) {
-                customRedisClient.set(key1, "0");
-            } else {
-                customRedisClient.decrement(key1, parallelism);
-            }
-            String key2 = "user:" + clusterUserId + ":using-license:carsim";
-            final String usingLicense2 = customRedisClient.get(key2);
-            if (StringUtil.isEmpty(usingLicense2)) {
-                customRedisClient.set(key2, "0");
-            } else {
-                customRedisClient.decrement(key2, parallelism);
-            }
+            releaseLicense("user:" + clusterUserId + ":using-license:simulation",parallelism);
+            releaseLicense("user:" + clusterUserId + ":using-license:dynamic",parallelism);
         } else {
             throw new RuntimeException("未知模型类型：" + modelType);
         }
     }
 
+    private void releaseLicense(String key, int number){
+        final String usingLicense = customRedisClient.get(key);
+        if (StringUtil.isEmpty(usingLicense)) {
+            customRedisClient.set(key, "0");
+        } else {
+            customRedisClient.decrement(key, number);
+        }
+    }
+
     public String getClusterUserIdByProjectUserId(String projectUserId) {
         String clusterUserId;
         final UserEntity userEntity = userMapper.selectById(projectUserId);