虚节点扫描线程重构

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

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.enums.RunRecordingEnum;
+import com.byit.model.JobTaskRunLog;
+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.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 日志的虚节点扫描
+ * 作用:扫描虚节点，去运行记录查询此虚节点对应的功能流是否已经完成
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class VirtualNodeThreadRunHelper extends BaseThreadRunHelper {
+
+    private static final String LOCK_NAME = "log_virtual_node_lock";
+    private final DataSource dataSource;
+
+    private final RunRecordingService runRecordingService;
+    private final JobTaskRunLogService jobTaskRunLogService;
+
+    public VirtualNodeThreadRunHelper(DataSource dataSource, RunRecordingService runRecordingService, JobTaskRunLogService jobTaskRunLogService) {
+        this.dataSource = dataSource;
+        this.runRecordingService = runRecordingService;
+        this.jobTaskRunLogService = jobTaskRunLogService;
+    }
+
+    @Override
+    public Long start() {
+
+        //查询日志表没有完结的虚拟节点
+        List<JobTaskRunLog> notEndVirtualNodes = jobTaskRunLogService.findNotEndVirtualNode( );
+        if(CollectionUtil.isNotEmpty(notEndVirtualNodes)){
+            notEndVirtualNodes.forEach(notEndVirtualNode ->{
+                log.debug("------------【开始查询虚拟节点的执行情况】----------------");
+                //根据运行标识和工作流id查询运行日志
+                RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getMapFlowId( ), notEndVirtualNode.getRunId());
+                //判断当前的工作流是否已经完结
+                if(runRecordingByFlowIdAndRunId!=null && RunRecordingEnum.FLOW_STATUS_IS_END.getCode().equals(runRecordingByFlowIdAndRunId.getFlowStatus())){
+                    log.debug("------------【查询到有已经完成的虚拟节点修改日志】----------------");
+                    notEndVirtualNode.setRunCode(runRecordingByFlowIdAndRunId.getFlowRunResult());
+                    notEndVirtualNode.setEndTime(new Date());
+                    //修改日志信息 改为成功或者失败
+                    jobTaskRunLogService.updateJobTaskRunLog(notEndVirtualNode);
+                }
+            });
+        }else{
+            return UNIVERSAL_WAIT_TIME;
+        }
+        return NOT_WAIT_TIME;
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+}