mlx

pom.xml

@@ -214,12 +214,20 @@
                 <artifactId>lombok</artifactId>
                 <version>${lombok.version}</version>
             </dependency>
-            <!-- junit -->
+            <!-- 测试 - 开始 -->
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-starter-test</artifactId>
+                <version>${spring-boot.version}</version>
+                <scope>test</scope>
+            </dependency>
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>
                 <version>${junit.version}</version>
+                <scope>test</scope>
             </dependency>
+            <!-- 测试 - 结束 -->
             <!-- log4j -->
             <dependency>
                 <groupId>log4j</groupId>

simulation-resource-scheduler/pom.xml

@@ -146,6 +146,7 @@
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>${spring-boot.version}</version>
                 <configuration>
                     <fork>true</fork>
                     <addResources>true</addResources>

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/configuration/redis/CustomRedisClient.java

@@ -1,12 +1,12 @@
 package com.css.simulation.resource.scheduler.configuration.redis;
 
-import api.common.util.CollectionUtil;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.util.Collection;
+import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -17,7 +17,7 @@ public class CustomRedisClient {
     @Resource
     private StringRedisTemplate stringRedisTemplate;
 
-    //* -------------------------------- Comment --------------------------------
+    //* -------------------------------- 基本操作 --------------------------------
 
     public Set<String> keys(String pattern) {
         return stringRedisTemplate.keys(pattern);
@@ -33,46 +33,58 @@ public class CustomRedisClient {
 
     public void flushdb() {
         Set<String> keys = stringRedisTemplate.keys("*");
-        if (CollectionUtil.isNotEmpty(keys)) {
+        if (keys != null && keys.size() != 0) {
             stringRedisTemplate.delete(keys);
         }
     }
 
-    public Boolean createLock(String key) {
-        return stringRedisTemplate.opsForValue().setIfAbsent(key, "", 1L, TimeUnit.HOURS);
-    }
-
-    public void deleteLock(String key) {
-        delete(key);
-    }
-
     public void delete(String key) {
         stringRedisTemplate.delete(key);
     }
 
-    //* -------------------------------- Comment --------------------------------
-
-    public void deleteByKey(String key) {
-        stringRedisTemplate.delete(key);
+    /**
+     * 加值
+     */
+    public void increment(String key, long number) {
+        stringRedisTemplate.opsForValue().increment(key, number);
     }
 
-    public void deleteByKeyCollection(Collection<String> keyCollection) {
-        stringRedisTemplate.delete(keyCollection);
+    /**
+     * 减值
+     */
+    public void decrement(String key, long number) {
+        stringRedisTemplate.opsForValue().decrement(key, number);
     }
 
-    public void deleteByPrefix(String prefix) {
-        Set<String> keySetByPrefix = getKeySetByPrefix(prefix);
-        deleteByKeyCollection(keySetByPrefix);
-    }
 
-    public String getStringByKey(String key) {
-        return stringRedisTemplate.opsForValue().get(key);
+    //* -------------------------------- 分布式锁 --------------------------------
+
+    /**
+     * 加锁，有阻塞
+     *
+     * @param expire  锁的过期时间，防止程序崩溃后无法释放锁
+     * @param timeout 间隔多久获取一次锁
+     */
+    @SneakyThrows
+    public void lock(String name, long expire, long timeout) {
+        Boolean success = false;
+        while (!success) {
+            success = tryLock(name, expire);
+            TimeUnit.MILLISECONDS.sleep(timeout);
+        }
     }
 
-
-    public Set<String> getKeySetByPrefix(String prefix) {
-        return stringRedisTemplate.keys(prefix + "*");
+    /**
+     * 尝试加锁，无阻塞
+     */
+    public Boolean tryLock(String name, long expire) {
+        return stringRedisTemplate.opsForValue().setIfAbsent(name, "None", Duration.ofMillis(expire));
     }
 
-
+    /**
+     * 解锁
+     */
+    public void unlock(String name) {
+        delete(name);
+    }
 }

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

@@ -1,5 +1,6 @@
 package com.css.simulation.resource.scheduler.service;
 
+import com.css.simulation.resource.scheduler.configuration.redis.CustomRedisClient;
 import com.css.simulation.resource.scheduler.entity.PrefixEntity;
 import com.css.simulation.resource.scheduler.entity.ProjectEntity;
 import com.css.simulation.resource.scheduler.entity.TaskEntity;
@@ -20,37 +21,41 @@ public class TaskService {
     private TaskMapper taskMapper;
     @Resource
     private ProjectUtil projectUtil;
+    @Resource
+    private CustomRedisClient customRedisClient;
+
 
     // -------------------------------- Comment --------------------------------
 
     @SneakyThrows
     public void taskState(String taskId, String state, String podName) {
-        log.info("接收到参数为：taskId={}，state={}，podName={}", taskId, state, podName);
-        TaskEntity taskEntity = taskMapper.selectById(taskId);
-        if (taskEntity == null) {
-            log.info("接收到已删除但还在执行的任务：taskId={}，state={}，podName={}", taskId, state, podName);
-            return;
-        }
-        String projectId = taskEntity.getPId(); // 项目 id
-        ProjectEntity projectEntity = projectUtil.getProjectByProjectId(projectId);
-        String projectType = projectEntity.getProjectType();  // 项目类型
-        String maxSimulationTime = projectEntity.getMaxSimulationTime();  // 项目类型
-        String userId = taskEntity.getCreateUserId();   // 用户 id
-        PrefixEntity redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaskId(userId, projectId, taskId);  // 项目前缀
-        log.info("判断项目是否已完成。");
-        boolean projectCompleted = taskManager.isProjectCompleted(redisPrefix, projectId, projectType, maxSimulationTime, taskId, state, podName);
-        if (!projectCompleted) {
-            return;
+        String lockName = "taskId:" + taskId + ":state:" + state + ":pod-name:" + podName;
+        customRedisClient.lock(lockName, 30 * 60L, 7);
+        try {
+            TaskEntity taskEntity = taskMapper.selectById(taskId);
+            String projectId = taskEntity.getPId(); // 项目 id
+            ProjectEntity projectEntity = projectUtil.getProjectByProjectId(projectId);
+            String projectType = projectEntity.getProjectType();  // 项目类型
+            String maxSimulationTime = projectEntity.getMaxSimulationTime();  // 项目类型
+            String userId = taskEntity.getCreateUserId();   // 用户 id
+            PrefixEntity redisPrefix = projectUtil.getRedisPrefixByUserIdAndProjectIdAndTaskId(userId, projectId, taskId);  // 项目前缀
+            log.info("判断项目 {} 是否已完成。", projectId);
+            boolean projectCompleted = taskManager.isProjectCompleted(redisPrefix, projectId, projectType, maxSimulationTime, taskId, state, podName);
+            if (!projectCompleted) {
+                return;
+            }
+            log.info("项目 {} 准备打分。", projectId);
+            taskManager.prepareScore(redisPrefix.getProjectRunningKey());
+            log.info("项目 {} 开始打分。", projectId);
+            taskManager.score(userId, projectId, projectType);
+            log.info("项目 {} 计算评价等级。", projectId);
+            taskManager.evaluationLevel(projectId);
+            log.info("项目 {} 开始释放资源。", projectId);
+            taskManager.done(redisPrefix, projectId, projectType);
+            log.info("项目 {} 运行结束。", projectId);
+        } finally {
+            customRedisClient.unlock(lockName);
         }
-        log.info("项目 {} 准备打分。", projectId);
-        taskManager.prepareScore(redisPrefix.getProjectRunningKey());
-        log.info("项目 {} 开始打分。", projectId);
-        taskManager.score(userId, projectId, projectType);
-        log.info("项目 {} 计算评价等级。", projectId);
-        taskManager.evaluationLevel(projectId);
-        log.info("项目 {} 开始释放资源。", projectId);
-        taskManager.done(redisPrefix, projectId, projectType);
-        log.info("项目 {} 运行结束。", projectId);
 
     }
 

simulation-resource-video/src/main/java/com/css/simulation/resource/video/configuration/redis/CustomRedisClient.java

@@ -1,12 +1,13 @@
 package com.css.simulation.resource.video.configuration.redis;
 
 import api.common.util.CollectionUtil;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
-import java.util.Collection;
+import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -38,41 +39,38 @@ public class CustomRedisClient {
         }
     }
 
-    public Boolean createLock(String key) {
-        return stringRedisTemplate.opsForValue().setIfAbsent(key, "", 1L, TimeUnit.HOURS);
-    }
-
-    public void deleteLock(String key) {
-        delete(key);
-    }
-
     public void delete(String key) {
         stringRedisTemplate.delete(key);
     }
 
-    //* -------------------------------- Comment --------------------------------
-
-    public void deleteByKey(String key) {
-        stringRedisTemplate.delete(key);
-    }
-
-    public void deleteByKeyCollection(Collection<String> keyCollection) {
-        stringRedisTemplate.delete(keyCollection);
-    }
-
-    public void deleteByPrefix(String prefix) {
-        Set<String> keySetByPrefix = getKeySetByPrefix(prefix);
-        deleteByKeyCollection(keySetByPrefix);
+    //* -------------------------------- 分布式锁 --------------------------------
+
+    /**
+     * 加锁，有阻塞
+     *
+     * @param expire  锁的过期时间，防止程序崩溃后无法释放锁
+     * @param timeout 间隔多久获取一次锁
+     */
+    @SneakyThrows
+    public void lock(String name, long expire, long timeout) {
+        Boolean success = false;
+        while (!success) {
+            success = tryLock(name, expire);
+            TimeUnit.MILLISECONDS.sleep(timeout);
+        }
     }
 
-    public String getStringByKey(String key) {
-        return stringRedisTemplate.opsForValue().get(key);
+    /**
+     * 尝试加锁，无阻塞
+     */
+    public Boolean tryLock(String name, long expire) {
+        return stringRedisTemplate.opsForValue().setIfAbsent(name, "None", Duration.ofMillis(expire));
     }
 
-
-    public Set<String> getKeySetByPrefix(String prefix) {
-        return stringRedisTemplate.keys(prefix + "*");
+    /**
+     * 解锁
+     */
+    public void unlock(String name) {
+        stringRedisTemplate.delete(name);
     }
-
-
 }