重构运行实例的扫描规则

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunRecordingServiceImpl.java

@@ -30,7 +30,7 @@ public class RunRecordingServiceImpl implements RunRecordingService {
 
     @Override
     public List<RunRecording> findRunningRunRecording() {
-        return null;
+        return runRecordingMapper.findRunningRunRecording();
     }
 
     @Override

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

@@ -73,8 +73,7 @@ public class LogScanHelper {
 
     public void start(){
         virtualNodeScanMethod();
-        //TODO  不推荐使用，详情见方法介绍
-        notAlarmedNodeScanMethod();
+        //notAlarmedNodeScanMethod();
         errorNodeScan();
     }
 

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

@@ -4,6 +4,8 @@ import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
 import com.byit.enums.RunRecordingEnum;
 import com.byit.filesystem.FileSystem;
+import com.byit.job.enums.JobResultEnum;
+import com.byit.job.exceptions.BusinessException;
 import com.byit.model.EmailAlarm;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.model.RunRecording;
@@ -12,6 +14,7 @@ import com.byit.service.JobTaskRunLogService;
 import com.byit.service.RunRecordingService;
 import com.byit.util.TimeFormatUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.csource.common.MyException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -105,8 +108,8 @@ public class RunRecordingScanHelper {
     public void judgeFlowIsEndThread(){
         judgeFlowIsEndThread = new Thread(() ->{
             dateAligned(5000);
-            log.info("----------------------【com.byit.thread.RunRecordingScanThread#judgeFlowIsEnd】-------------------");
-            while (judgeFlowIsEndStop){
+            log.info("----------------------【com.byit.thread.RunRecordingScanThread#judgeFlowIsEnd】start-------------------");
+            while (!judgeFlowIsEndStop){
                 boolean isSleep = false;
                 Connection conn = null;
                 Boolean connAutoCommit = null;
@@ -118,7 +121,8 @@ public class RunRecordingScanHelper {
                     preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'judge_flow_end_lock' FOR UPDATE ");
                     preparedStatement.execute();
                     //进行操作
-
+                    judgeFlowIsEnd();
+                    isSleep = true;
                 }catch (Exception e){
                     if (!judgeFlowIsEndStop) {
                         e.printStackTrace();
@@ -166,18 +170,73 @@ public class RunRecordingScanHelper {
                 }
 
                 if(isSleep){
-                    dateAligned(10000);
+                    dateAligned(20000);
                 }
             }
         });
+
+        judgeFlowIsEndThread.setDaemon(true);
+        judgeFlowIsEndThread.setName("myth-job#【judgeFlowIsEndThread】# judgeFlowIsEndThread");
+        judgeFlowIsEndThread.start();
     }
 
     /**
      * 扫描未完成的工作流
      */
     private void judgeFlowIsEnd(){
+        log.debug("---------------开始扫描运行中的运行实例----------------");
         //查询未完结的工作流实例
+        List<RunRecording> runningRunRecordings = runRecordingService.findRunningRunRecording();
+        log.debug("---------扫描到运行实例{}个----------",runningRunRecordings.size());
+        runningRunRecordings.forEach(runRecording -> {
+            String runId = runRecording.getRunId();
+            Integer flowId = runRecording.getFlowId();
+            List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBsByFlowIdAndRunId = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
+            //判断节点数量与设定数量是否一致
+            if(jobTaskRunLogWithBLOBsByFlowIdAndRunId==null || jobTaskRunLogWithBLOBsByFlowIdAndRunId.size() != runRecording.getFlowNodeCount()){
+                log.debug("------------发现工作流{}不符合条件----------------",jobTaskRunLogWithBLOBsByFlowIdAndRunId);
+                return;
+            }
+
+            //判断是否全部完结  如果全部完结则判断工作流是否成功
+            if(logIsAllEnd(jobTaskRunLogWithBLOBsByFlowIdAndRunId)){
+
+                String runCode = allNodeIsSuccess(jobTaskRunLogWithBLOBsByFlowIdAndRunId);
+                //设置运行结果
+                runRecording.setFlowRunResult(runCode);
+                log.debug("----------------扫描到有完结的运行实例{},{}-------------",runCode,RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
+                runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
+                runRecordingService.updateRunRecordingById(runRecording);
+            }
+        });
+    }
 
+    private String allNodeIsSuccess(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
+        for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
+            if (StringUtils.isNotEmpty(jobTaskRunLogWithBLOB.getRunCode())) {
+                if(!"1".equals(jobTaskRunLogWithBLOB.getRunCode()) || !"3".equals(jobTaskRunLogWithBLOB.getRunCode())){
+                    return jobTaskRunLogWithBLOB.getRunCode();
+                }
+            }else{
+                throw new BusinessException(JobResultEnum.RUN_MSG_FAIL);
+            }
+        }
+        return "1";
+    }
+
+    /**
+     * 判断节点是否全部完结
+     * @param jobTaskRunLogWithBLOBs
+     * @return
+     */
+    private boolean logIsAllEnd(List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBLOBs){
+        for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOB : jobTaskRunLogWithBLOBs) {
+            if ("0".equals(jobTaskRunLogWithBLOB.getRunCount())) {
+                //存在运行中的直接返回false
+                return false;
+            }
+        }
+        return true;
     }
 
 
@@ -313,6 +372,23 @@ public class RunRecordingScanHelper {
             }
         }
         log.warn("---------------【运行记录扫描日志线程被注销】-----------------------");
+
+        this.judgeFlowIsEndStop = true;
+        try {
+            TimeUnit.SECONDS.sleep(1);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        if (judgeFlowIsEndThread.getState() != Thread.State.TERMINATED) {
+            judgeFlowIsEndThread.interrupt();
+
+            try {
+                judgeFlowIsEndThread.join();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        log.warn("---------------【扫描工作流结果线程被注销】-----------------------");
     }
 
     /**

byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml

@@ -36,7 +36,7 @@
     from run_recording
     where flow_status ='4' and is_alarm = '1'
   </select>
-
+  <!--查询运行中的-->
   <select id="findRunningRunRecording" resultMap="BaseResultMap">
     select
         <include refid="Base_Column_List" />

byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/JobResultEnum.java

@@ -9,7 +9,9 @@ public enum JobResultEnum implements IEnum {
     FAIL("100500","任务执行失败","2"),
     FAIL_TIMEOUT("100502","超时错误","2"),
     DISPATCH_SUCCESS("200200","调度成功","1"),
-    DISPATCH_FAIL("200500","调度失败","2");
+    DISPATCH_FAIL("200500","调度失败","2"),
+
+    RUN_MSG_FAIL("300000","执行结果异常","2");
     private String code;
     private String msg;
     private String res;