任务实例超时快速失败功能

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskMapper.java

@@ -90,4 +90,5 @@ public interface JobTaskMapper {
      * @return
      */
     int startByRunId(String runId);
+
 }
\ No newline at end of file

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

@@ -52,5 +52,7 @@ public interface JobTaskService {
      */
     void removeMythJobTaskInIds(List<JobTaskSchedule> jobTasks);
 
-    void deleteInIds(@Param("ids") List<Integer> ids);
+    void deleteInIds(List<Integer> ids);
+
+    void removeByRunId(String runId);
 }

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

@@ -79,4 +79,9 @@ public class JobTaskServiceImpl implements JobTaskService {
     public void deleteInIds(List<Integer> ids) {
         jobTaskMapper.deleteInIds(ids);
     }
+
+    @Override
+    public void removeByRunId(String runId) {
+        jobTaskMapper.deleteByRunId(runId);
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/RunRecordingAndLogService.java

+package com.byit.service.mapservice;
+
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.model.NodeVersion;
+import com.byit.model.RunRecording;
+
+import java.util.List;
+
+/**
+ * @author huangfu
+ */
+public interface RunRecordingAndLogService {
+    /**
+     * 快速失败
+     * @param runRecording 当前这个超时的实例
+     */
+    void logAndRunRecordingFailFast(RunRecording runRecording);
+}

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

+package com.byit.service.mapservice.impl;
+
+import com.byit.enums.RunRecordingEnum;
+import com.byit.enums.task.RunResultEnum;
+import com.byit.model.JobTaskRunLog;
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.model.NodeVersion;
+import com.byit.model.RunRecording;
+import com.byit.service.JobTaskRunLogService;
+import com.byit.service.JobTaskService;
+import com.byit.service.NodeVersionService;
+import com.byit.service.RunRecordingService;
+import com.byit.service.mapservice.RunRecordingAndLogService;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Service;
+
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 快速失败
+ * @author huangfu
+ */
+@Service
+public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService {
+    public static final String TRIGGER_MSG = "任务执行超时，被强制执行快速失败！";
+    private final JobTaskRunLogService jobTaskRunLogService;
+    private final RunRecordingService runRecordingService;
+    private final NodeVersionService nodeVersionService;
+    private final JobTaskService jobTaskService;
+
+    public RunRecordingAndLogServiceImpl(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService,
+                                         NodeVersionService nodeVersionService, JobTaskService jobTaskService) {
+        this.jobTaskRunLogService = jobTaskRunLogService;
+        this.runRecordingService = runRecordingService;
+        this.nodeVersionService = nodeVersionService;
+        this.jobTaskService = jobTaskService;
+    }
+
+    @Override
+    public void logAndRunRecordingFailFast(RunRecording runRecording) {
+        String runId = runRecording.getRunId();
+        Integer flowId = runRecording.getFlowId();
+        //查询该实例对应的所由日志节点
+        List<JobTaskRunLogWithBLOBs> jobTaskRunLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
+        List<Integer> logNodeId = jobTaskRunLog.stream().map(JobTaskRunLog::getNodeId).collect(Collectors.toList());
+        List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(flowId);
+        //筛选没在日志里面的节点
+        List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
+        //将这些节点置为失败并将实例也置为失败
+        //保存错误日志节点
+        notLogNode.forEach(node ->{
+            JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
+            BeanUtils.copyProperties(node,log);
+            log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
+            log.setTriggerMsg(TRIGGER_MSG);
+            log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
+            log.setTriggerMsg(TRIGGER_MSG);
+            Date thisTime = new Date();
+            log.setStartTime(thisTime);
+            log.setEndTime(thisTime);
+            log.setTriggerTime(thisTime);
+            jobTaskRunLogService.saveJobTaskRunLog(log);
+        });
+        //执行实例的快速失败
+        if (runRecording.getStartTime() == null) {
+            runRecording.setStartTime(new Date());
+        }
+        runRecording.setEndTime(new Date());
+        runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
+        runRecording.setFlowRunResult(RunRecordingEnum.RUN_FLOW_FAILURE.getCode());
+        runRecording.setFailFast(RunRecordingEnum.FAIL_FAST_YES.getCode());
+        runRecordingService.updateRunRecordingById(runRecording);
+        //删除task里面的数据
+        jobTaskService.removeByRunId(runId);
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java

@@ -5,10 +5,7 @@ import com.byit.conf.MythJobAutoConfigure;
 import com.byit.dto.executor.DispatchResponseDto;
 import com.byit.dto.executor.ScriptDto;
 import com.byit.dto.plugin.RunLog;
-import com.byit.enums.EmailEnum;
-import com.byit.enums.JobResultEnum;
-import com.byit.enums.NodePropertyEnum;
-import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.*;
 import com.byit.enums.task.RunResultEnum;
 import com.byit.enums.task.RunTypeEnum;
 import com.byit.model.JobTaskRunLogWithBLOBs;
@@ -49,9 +46,10 @@ public class ScriptExecutorJobTask implements TimerTask {
     @Override
     public void run(Timeout timeout) {
         log.debug("---------开始交验工作流时否正在运行中------------");
-        if(checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(),mythJobTaskSchedule.getRunId())){
-           log.warn("--------------该工作流已经被杀死，执行快速失败！-------------------");
-            saveErrorLog(mythJobTaskSchedule);
+        String code = checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(), mythJobTaskSchedule.getRunId());
+        if(code != null){
+           log.warn("--------------该工作流已经被杀死快速失败，执行快速失败！-------------------");
+            saveErrorLog(mythJobTaskSchedule,code);
            return;
         }
         log.debug("-----------------工作流校验完成-------------");
@@ -68,18 +66,23 @@ public class ScriptExecutorJobTask implements TimerTask {
     }
 
     /**
-     * 校验工作流是否被杀死
+     * 校验工作流是否被杀死或者被快速失败
      * @param flowId 工作流ID
      * @param runId 运行标识
      * @return
      */
-    private boolean checkFlowStatusIsKill(Integer flowId,String runId){
+    private String checkFlowStatusIsKill(Integer flowId,String runId){
         RunRecordingServiceImpl bean = SpringUtil.getBean(RunRecordingServiceImpl.class);
         RunRecording runRecordingByFlowIdAndRunId = bean.findRunRecordingByFlowIdAndRunId(flowId, runId);
         if(runRecordingByFlowIdAndRunId != null){
-            return "5".equals(runRecordingByFlowIdAndRunId.getFlowRunResult());
+            if (RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(runRecordingByFlowIdAndRunId.getFlowRunResult())) {
+                return RunRecordingEnum.RUN_FLOW_KILL.getCode();
             }
-        return false;
+            if (RunRecordingEnum.FAIL_FAST_YES.getCode().equals(runRecordingByFlowIdAndRunId.getFailFast())) {
+                return RunRecordingEnum.FAIL_FAST_YES.getCode();
+            }
+        }
+        return null;
     }
 
     /**
@@ -168,8 +171,19 @@ public class ScriptExecutorJobTask implements TimerTask {
      * 保存失败的节点执行信息
      * @param mythJobTaskSchedule 节点信息
      */
-    private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule){
+    private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule,String code){
         log.debug("-----------saveErrorLog--保存KILL脚本调度日志开始------------");
+        String msg = "";
+        if(RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(code)){
+            msg = RunRecordingEnum.RUN_FLOW_KILL.getMessage();
+        }
+        if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)) {
+            msg = RunRecordingEnum.FAIL_FAST_YES.getMessage();
+        }
+
+        if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)){
+            code = RunResultEnum.RUN_ERROR.getCode();
+        }
         JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
         //获取值放入调度轮之前保存的日志
         JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
@@ -178,9 +192,9 @@ public class ScriptExecutorJobTask implements TimerTask {
         if (jobTaskRunLogById.getRunCount()>1) {
             //上一次调度的日志
             String triggerMsg = jobTaskRunLogById.getTriggerMsg();
-            jobTaskRunLog.setTriggerMsg(triggerMsg+RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
+            jobTaskRunLog.setTriggerMsg(triggerMsg+msg+LINE);
         }else{
-            jobTaskRunLog.setTriggerMsg(RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
+            jobTaskRunLog.setTriggerMsg(msg+LINE);
         }
 
         jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
@@ -197,14 +211,14 @@ public class ScriptExecutorJobTask implements TimerTask {
         //将剩余失败重试次数刷新为0
         jobTaskRunLog.setFailedRemainingCount(0);
         //将执行状态更改为被杀死
-        jobTaskRunLog.setRunCode(RunResultEnum.KILL_SUCCESS.getCode());
+        jobTaskRunLog.setRunCode(code);
         jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
         if (jobTaskRunLogById.getRunCount()>1) {
             //上一次的执行日志
             String runMsg = jobTaskRunLogById.getRunMsg();
-            jobTaskRunLog.setRunMsg(runMsg+RunResultEnum.KILL_SUCCESS.getMsg()+LINE);
+            jobTaskRunLog.setRunMsg(runMsg+msg+LINE);
         }else{
-            jobTaskRunLog.setRunMsg(RunResultEnum.KILL_SUCCESS.getMsg()+ LINE);
+            jobTaskRunLog.setRunMsg(msg+ LINE);
         }
         jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
         jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);

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

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.dto.FlowConditionDto;
+import com.byit.model.RunRecording;
+import com.byit.service.RunRecordingService;
+import com.byit.service.mapservice.RunRecordingAndLogService;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * 扫描超时线程 执行工作流的快速失败
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class TimeoutExampleThreadRunHelper extends BaseThreadRunHelper {
+
+    public static final String TIMEOUT_LOCK = "timeout_lock";
+    private final DataSource dataSource;
+    private final RunRecordingService runRecordingService;
+    private final RunRecordingAndLogService runRecordingAndLogService;
+
+    public TimeoutExampleThreadRunHelper(DataSource dataSource, RunRecordingService runRecordingService, RunRecordingAndLogService runRecordingAndLogService) {
+        this.dataSource = dataSource;
+        this.runRecordingService = runRecordingService;
+        this.runRecordingAndLogService = runRecordingAndLogService;
+    }
+
+    @Override
+    public Long start() {
+        List<RunRecording> allRunIng = runRecordingService.findAllRunIng(new FlowConditionDto());
+
+        List<RunRecording> timeoutRunRecordings = allRunIng.stream().filter(runRecording -> {
+            long startTime = runRecording.getStartTime().getTime();
+            long thisTime = System.currentTimeMillis();
+            Long flowTimeout = runRecording.getFlowTimeout();
+            long time = thisTime - startTime;
+            return time > flowTimeout;
+        }).collect(Collectors.toList());
+
+        if (CollectionUtil.isNotEmpty(timeoutRunRecordings)){
+            timeoutRunRecordings.forEach(timeoutRunRecording ->{
+
+                runRecordingAndLogService.logAndRunRecordingFailFast(timeoutRunRecording);
+            });
+        }
+        return TimeUnit.MINUTES.toMillis(5);
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return TIMEOUT_LOCK;
+    }
+}

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

@@ -469,6 +469,7 @@
 
   <delete id="deleteByRunId" parameterType="java.lang.Integer">
     delete from job_task
-    where run_id = #{runId,jdbcType=INTEGER}
+    where run_id = #{runId,jdbcType=VARCHAR}
   </delete>
+
 </mapper>
\ No newline at end of file