优化代码

fanxing-task-scheduling/src/main/java/com/viontech/fanxing/task/scheduling/controller/TaskController.java

@@ -10,7 +10,6 @@ import com.viontech.fanxing.task.scheduling.service.TaskService;
 import com.viontech.fanxing.task.scheduling.service.VAServerService;
 import com.viontech.keliu.util.JsonMessageUtil;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.springframework.web.bind.annotation.*;
 
 import javax.annotation.Resource;
@@ -47,20 +46,8 @@ public class TaskController implements TaskSchedulingTasksAdapter {
         taskData.setStoreConfig(storeConfigVo.getContent());
 
         // 计算运行时间并生成任务
-        String unid = task.getUnid();
-        ImmutablePair<Long, Long> nextTime = taskData.getRuntimeConfig().getNextTimeOfExecutionAndTerminal();
-        Long nextExecuteTime = nextTime.left;
-        Long nextTerminateTime = nextTime.right;
-        if (nextExecuteTime != null) {
-            taskService.addExecutableTaskData(unid, nextExecuteTime);
-            if (nextTerminateTime != null) {
-                taskService.addTerminatableTaskData(unid, nextTerminateTime);
-            }
-            taskService.addTaskData(taskData);
-            return JsonMessageUtil.getSuccessJsonMsg("success");
-        } else {
-            return JsonMessageUtil.getErrorJsonMsg("任务找不到可执行时间");
-        }
+        boolean success = taskService.distributeTask(taskData);
+        return success ? JsonMessageUtil.getSuccessJsonMsg("success") : JsonMessageUtil.getErrorJsonMsg("任务找不到可执行时间");
     }
 
     @Override

fanxing-task-scheduling/src/main/java/com/viontech/fanxing/task/scheduling/service/RedisCacheService.java

@@ -3,14 +3,11 @@ package com.viontech.fanxing.task.scheduling.service;
 import com.viontech.fanxing.commons.constant.RedisKeys;
 import com.viontech.fanxing.task.scheduling.model.TaskData;
 import com.viontech.fanxing.task.scheduling.model.vaserver.VaServerInfo;
-import org.redisson.api.RBucket;
-import org.redisson.api.RMap;
-import org.redisson.api.RScoredSortedSet;
-import org.redisson.api.RedissonClient;
+import org.redisson.api.*;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
-import java.util.Date;
+import java.util.concurrent.TimeUnit;
 
 /**
  * .
@@ -50,4 +47,16 @@ public class RedisCacheService {
         return redissonClient.getBucket(key);
     }
 
+    public RLock getLockMust(String lockName) {
+        RLock lock = redissonClient.getLock(lockName);
+        boolean isLock = false;
+        while (!isLock) {
+            try {
+                isLock = lock.tryLock(30, 25, TimeUnit.SECONDS);
+            } catch (InterruptedException ignore) {
+            }
+        }
+        return lock;
+    }
+
 }

fanxing-task-scheduling/src/main/java/com/viontech/fanxing/task/scheduling/service/TaskService.java

 package com.viontech.fanxing.task.scheduling.service;
 
 import com.viontech.fanxing.commons.model.Task;
+import com.viontech.fanxing.task.scheduling.model.RuntimeConfig;
 import com.viontech.fanxing.task.scheduling.model.TaskData;
 import com.viontech.fanxing.task.scheduling.model.vaserver.VaServerInfo;
 import org.apache.commons.lang3.tuple.ImmutablePair;
 import org.redisson.api.RMap;
 import org.redisson.api.RScoredSortedSet;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import tech.powerjob.client.PowerJobClient;
+import tech.powerjob.common.request.http.SaveJobInfoRequest;
 
 import javax.annotation.Resource;
 
@@ -24,14 +28,23 @@ public class TaskService {
     @Resource
     private VAServerService vaServerService;
 
-    public boolean addExecutableTaskData(String taskUnid, Long timestamp) {
-        RScoredSortedSet<String> toBeExecutedTaskUnidSet = redisCacheService.getToBeExecutedTaskUnidSet();
-        return toBeExecutedTaskUnidSet.add(timestamp, taskUnid);
-    }
-
-    public boolean addTerminatableTaskData(String taskUnid, Long timestamp) {
-        RScoredSortedSet<String> toBeTerminatedTaskUnidSet = redisCacheService.getToBeTerminatedTaskUnidSet();
-        return toBeTerminatedTaskUnidSet.add(timestamp, taskUnid);
+    public boolean distributeTask(TaskData taskData) {
+        RuntimeConfig runtimeConfig = taskData.getRuntimeConfig();
+        String taskUnid = taskData.getTask().getUnid();
+        ImmutablePair<Long, Long> nextTime = runtimeConfig.getNextTimeOfExecutionAndTerminal();
+        Long nextExecuteTime = nextTime.left;
+        Long nextTerminateTime = nextTime.right;
+        if (nextExecuteTime != null) {
+            RScoredSortedSet<String> toBeExecutedTaskUnidSet = redisCacheService.getToBeExecutedTaskUnidSet();
+            toBeExecutedTaskUnidSet.add(nextExecuteTime, taskUnid);
+            if (nextTerminateTime != null) {
+                RScoredSortedSet<String> toBeTerminatedTaskUnidSet = redisCacheService.getToBeTerminatedTaskUnidSet();
+                toBeTerminatedTaskUnidSet.add(nextTerminateTime, taskUnid);
+            }
+            return true;
+        } else {
+            return false;
+        }
     }
 
     public void addTaskData(TaskData taskData) {

fanxing-task-scheduling/src/main/java/com/viontech/fanxing/task/scheduling/service/VAServerService.java

@@ -6,6 +6,7 @@ import com.viontech.fanxing.commons.model.Task;
 import com.viontech.fanxing.task.scheduling.model.TaskData;
 import com.viontech.fanxing.task.scheduling.model.vaserver.VaServerInfo;
 import org.redisson.api.RBucket;
+import org.redisson.api.RLock;
 import org.redisson.api.RMap;
 import org.springframework.stereotype.Service;
 
@@ -87,7 +88,7 @@ public class VAServerService {
         server.setAvailableResources(server.getAvailableResources() - task.getResourceNeed());
         RMap<String, VaServerInfo> vaServerMap = redisCacheService.getVaServerMap();
         vaServerMap.put(server.getDevID(), server);
-        return false;
+        return true;
     }
 
     /**
@@ -112,8 +113,14 @@ public class VAServerService {
             vaServerHttpService.rmTask(taskUnid, vaServerInfo);
             // 解除任务和 vaServer 关联, 恢复资源数量
             map.remove(taskUnid);
-            vaServerInfo.setAvailableResources(vaServerInfo.getAvailableResources() + task.getResourceNeed());
-            vaServerMap.put(vaServerId, vaServerInfo);
+
+            RLock vaServerLock = redisCacheService.getLockMust("lock:vaserver:" + vaServerId);
+            try {
+                vaServerInfo.setAvailableResources(vaServerInfo.getAvailableResources() + task.getResourceNeed());
+                vaServerMap.put(vaServerId, vaServerInfo);
+            } finally {
+                vaServerLock.forceUnlock();
+            }
             return true;
         }
         return true;