对重跑节点的单独处理

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/NodeNameEnum.java

@@ -7,6 +7,7 @@ package com.byit.enums;
 
 public enum NodeNameEnum {
     START_NODE("start","开始节点"),
+    END_NODE("end","结束节点"),
     ;
     private String nodeName;
     private String details;

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/SpecialFlowThreadRunHelper.java

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.enums.FlowPropertyEnum;
+import com.byit.enums.ScheduleEnum;
+import com.byit.model.JobTask;
+import com.byit.model.JobTaskSchedule;
+import com.byit.service.JobTaskService;
+import com.byit.service.mapservice.RunRecordingAndJobTaskService;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.sql.DataSource;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 特殊情况的工作流扫描
+ * @author huangfu
+ */
+@Slf4j
+public class SpecialFlowThreadRunHelper extends BaseThreadRunHelper {
+    /**
+     * 读取任务节点的预读
+     */
+    private static final long PRE_READ_MS = 7000;
+    private static final String LOCK_NAME = "special_flow_lock";
+    private final DataSource dataSource;
+    private final JobTaskService jobTaskService;
+    /**
+     * 运行实例和task的组合操作
+     */
+    private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
+
+    public SpecialFlowThreadRunHelper(DataSource dataSource, JobTaskService jobTaskService, RunRecordingAndJobTaskService runRecordingAndJobTaskService) {
+        this.dataSource = dataSource;
+        this.jobTaskService = jobTaskService;
+        this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
+    }
+
+    @Override
+    public Long start() {
+        long nowTime = System.currentTimeMillis();
+        //开始寻找此时 不是暂停状态，而且七秒内即将运行的任务 而且还不是暂停的节点
+        List<JobTask> repairs = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS, ScheduleEnum.REPAIR.getCode());
+        List<JobTask> repeats = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS, ScheduleEnum.REPEAT.getCode());
+
+        List<JobTask> jobTasks = new ArrayList<>(8);
+        if(CollectionUtil.isNotEmpty(repairs)){
+            jobTasks.addAll(repairs);
+        }
+        if(CollectionUtil.isNotEmpty(repeats)){
+            jobTasks.addAll(repeats);
+        }
+        if(CollectionUtil.isNotEmpty(jobTasks)) {
+            List<JobTaskSchedule> jobTaskSchedules = new ArrayList<>(15);
+            for (JobTask jobTask : jobTasks) {
+                //获取上级节点
+                String nodeDepend = jobTask.getNodeDepend();
+
+                if(FlowPropertyEnum.IS_INNER.getCode().equals(jobTask.getIsVirtual())) {
+                    //虚节点处理操作
+                    innerNodeOperating(jobTask);
+                }else{
+                    //判断上级节点是否存在  不存在  直接跑，存在则进行判断
+                    if(StringUtils.isNotBlank(nodeDepend)){
+                        String[] split = nodeDepend.split(",");
+                    }else{
+
+                    }
+                }
+
+
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 内嵌节点处理操作
+     * @param thisJobTask 当前的任务节点
+     */
+    private void innerNodeOperating(JobTask thisJobTask){
+        try {
+            runRecordingAndJobTaskService.saveRunRecordingAndTask(thisJobTask);
+
+        } catch (Exception e) {
+            log.error("--------------------虚节点处理出现异常{}------------------",e.getMessage());
+        }
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java

 package com.byit.thread.helper;
 
 import cn.hutool.core.collection.CollectionUtil;
-import com.byit.enums.FlowPropertyEnum;
-import com.byit.enums.NodeNameEnum;
-import com.byit.enums.NodePropertyEnum;
-import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.*;
 import com.byit.job.exceptions.BusinessException;
 import com.byit.model.JobTask;
 import com.byit.model.JobTaskRunLog;
@@ -140,8 +137,15 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
             }
         }
 
+        String runId;
+        if(ScheduleEnum.REPEAT.getCode() .equals(thisJobTask.getScheduleType()) ){
+            runId = thisJobTask.getReRunId();
+        }else{
+            runId = thisJobTask.getRunId();
+        }
+
         //这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
-        List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(dependIdByNodeId, thisJobTask.getRunId());
+        List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(dependIdByNodeId, runId);
         if (CollectionUtil.isNotEmpty(jobTaskRunLogList)) {
             if(CollectionUtil.isEmpty(dependIdByNodeId)){
                 throw new BusinessException(NodeRunStatusPropertyEnum.NODE_RELY_ERROR.getMsg());