运行实例线程重构

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

+package com.byit.thread.helper;
+
+import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.RunRecordingEnum;
+import com.byit.job.enums.JobResultEnum;
+import com.byit.job.exceptions.BusinessException;
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.model.RunRecording;
+import com.byit.service.JobTaskRunLogService;
+import com.byit.service.RunRecordingService;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+
+/**
+ * 完结实例的线程
+ * 作用: 查询运行中的工作流实例，查看是否完成
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper {
+    private static final String LOCK_NAME = "judge_flow_end_lock";
+    private final DataSource dataSource;
+
+    private final JobTaskRunLogService jobTaskRunLogService;
+    private final RunRecordingService runRecordingService;
+
+
+
+    public ClosingExampleThreadRunHelper(DataSource dataSource, JobTaskRunLogService jobTaskRunLogService,
+                                         RunRecordingService runRecordingService) {
+        this.dataSource = dataSource;
+        this.jobTaskRunLogService = jobTaskRunLogService;
+        this.runRecordingService = runRecordingService;
+    }
+
+    @Override
+    public Long start() {
+        log.debug("---------------开始扫描运行中的运行实例----------------");
+        //查询未完结的工作流实例
+        List<RunRecording> runningRunRecordings = runRecordingService.findRunningRunRecording();
+        log.debug("---------扫描到运行实例{}个----------",runningRunRecordings.size());
+        runningRunRecordings.forEach(runRecording -> {
+            String runId = runRecording.getRunId();
+            Integer flowId = runRecording.getFlowId();
+            List<JobTaskRunLogWithBLOBs> jobTaskRunLogWithBs = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flowId, runId);
+            //判断节点数量与设定数量是否一致
+            if(jobTaskRunLogWithBs==null || jobTaskRunLogWithBs.size() != runRecording.getFlowNodeCount()){
+                log.debug("------------发现工作流{}不符合条件----------------",jobTaskRunLogWithBs);
+                return;
+            }
+
+            //判断是否全部完结  如果全部完结则判断工作流是否成功
+            if(logIsAllEnd(jobTaskRunLogWithBs)){
+
+                String runCode = allNodeIsSuccess(jobTaskRunLogWithBs);
+                //设置运行结果
+                runRecording.setFlowRunResult(runCode);
+                log.debug("----------------扫描到有完结的运行实例{},{}-------------",runCode, RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
+                runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
+                runRecordingService.updateRunRecordingById(runRecording);
+            }
+        });
+        return UNIVERSAL_WAIT_TIME;
+    }
+
+    /**
+     * 判断实例的执行状态
+     * @param jobTaskRunLogs 节点信息
+     * @return 实例的结果
+     */
+    private String allNodeIsSuccess(List<JobTaskRunLogWithBLOBs> jobTaskRunLogs){
+        for (JobTaskRunLogWithBLOBs jobTaskRunLog : jobTaskRunLogs) {
+            if (StringUtils.isNotEmpty(jobTaskRunLog.getRunCode())) {
+                if(!NodeRunStatusPropertyEnum.RUN_SUCCESS.getCode().equals(jobTaskRunLog.getRunCode())
+                        || !NodeRunStatusPropertyEnum.RE_RUN_SUCCESS.getCode().equals(jobTaskRunLog.getRunCode())){
+                    return jobTaskRunLog.getRunCode();
+                }
+            }else{
+                throw new BusinessException(JobResultEnum.RUN_MSG_FAIL);
+            }
+        }
+        return NodeRunStatusPropertyEnum.RUN_SUCCESS.getCode();
+    }
+
+    /**
+     * 判断节点是否全部完结
+     * @param jobTaskRunLogs 此实例全部的节点 运行信息
+     * @return 是否全部完结
+     */
+    private boolean logIsAllEnd(List<JobTaskRunLogWithBLOBs> jobTaskRunLogs){
+        for (JobTaskRunLogWithBLOBs jobTaskRunLog : jobTaskRunLogs) {
+            if (NodeRunStatusPropertyEnum.RUN_ING.getCode().equals(jobTaskRunLog.getRunCode())) {
+                log.debug("--------------节点{}，正在运行中-----------",jobTaskRunLog);
+                //存在运行中的直接返回false
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @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/EndAndNotWarningThreadRunHelper.java

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.enums.RunRecordingEnum;
+import com.byit.model.RunRecording;
+import com.byit.service.RunRecordingService;
+import com.byit.service.mapservice.RunRecordingAndEmailService;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+
+/**
+ * 查询完结且没有告警的节点
+ * 作用: 查询运行实例里面 完结而且没有告警的实例，根据告警规则添加进邮箱表，交由邮箱管理
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
+
+    private static final String LOCK_NAME = "run_recording_lock";
+    private final DataSource dataSource;
+
+
+    /**
+     * 完成时告警
+     */
+    public static final String WHEN_DONE = "1";
+    /**
+     * 失败时告警
+     */
+    public static final String FAILURE_DONE = "2";
+    /**
+     * 成功时告警
+     */
+    public static final String SUCCESS_DONE = "3";
+    /**
+     * 运行实例操作
+     */
+    private final RunRecordingService runRecordingService;
+    private final RunRecordingAndEmailService runRecordingAndEmailService;
+
+
+
+    public EndAndNotWarningThreadRunHelper(DataSource dataSource, RunRecordingService runRecordingService, RunRecordingAndEmailService runRecordingAndEmailService) {
+        this.dataSource = dataSource;
+        this.runRecordingService = runRecordingService;
+        this.runRecordingAndEmailService = runRecordingAndEmailService;
+    }
+
+    @Override
+    public Long start() {
+        //查询完结且未告警的工作流信息
+        List<RunRecording> runRecordingByEndAndNotIsAlarm = runRecordingService.findRunRecordingByEndAndNotIsAlarm();
+        if (CollectionUtil.isNotEmpty(runRecordingByEndAndNotIsAlarm)) {
+            runRecordingByEndAndNotIsAlarm.forEach(runRecording -> {
+                //如果设置为完成时告警
+                switch (runRecording.getAlarmlAction()) {
+                    //设置为完成时告警
+                    case WHEN_DONE:
+                        log.info("------工作流{},被设置为完成时告警-----",runRecording);
+                        if (RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecording.getFlowStatus())) {
+                            runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
+                        }
+                        break;
+                    //失败时告警
+                    case FAILURE_DONE:
+                        log.info("------工作流{},被设置为失败时告警-----",runRecording);
+                        if (RunRecordingEnum.RUN_FLOW_FAILURE.getCode().equals(runRecording.getFlowRunResult())
+                                || RunRecordingEnum.RUN_FLOW_RE_FAILURE.getCode().equals(runRecording.getFlowRunResult())) {
+                            runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
+                        }
+                        break;
+                    //成功时告警
+                    case SUCCESS_DONE:
+                        log.info("------工作流{},被设置为成功时告警-----",runRecording);
+                        if (RunRecordingEnum.RUN_FLOW_SUCCESS.getCode().equals(runRecording.getFlowRunResult())
+                                || RunRecordingEnum.RUN_FLOW_RE_SUCCESS.getCode().equals(runRecording.getFlowRunResult())) {
+                            runRecordingAndEmailService.saveEmailAndRunRecording(runRecording);
+                        }
+                        break;
+                    default:
+                        log.info("------工作流{},告警类别未知-----",runRecording);
+                        break;
+                }
+            });
+        }else{
+            return UNIVERSAL_WAIT_TIME;
+        }
+        return NOT_WAIT_TIME;
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+}