工作流扫描线程修改

byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java

@@ -25,17 +25,18 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
     private final JobScheduleHelper jobScheduleHelper;
     private final LogScanHelper logScanHelper;
     private final RunRecordingScanHelper runRecordingScanHelper;
-    //private final EmailScanHelper emailScanHelper;
-    private final FlowScanHelper flowScanHelper;
+    /**
+     * private final EmailScanHelper emailScanHelper
+     * private final FlowScanHelper flowScanHelper
+     */
     private final ApplicationContext applicationContext;
 
 
     @Autowired
-    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper, FlowScanHelper flowScanHelper, ApplicationContext applicationContext) {
+    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper, ApplicationContext applicationContext) {
         this.jobScheduleHelper = jobScheduleHelper;
         this.logScanHelper = logScanHelper;
         this.runRecordingScanHelper = runRecordingScanHelper;
-        this.flowScanHelper = flowScanHelper;
         this.applicationContext = applicationContext;
     }
 
@@ -49,7 +50,6 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
         this.jobScheduleHelper.doStop();
         this.logScanHelper.doStop();
         this.runRecordingScanHelper.doStop();
-        this.flowScanHelper.doStop();
     }
 
     /**
@@ -64,7 +64,6 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
         this.jobScheduleHelper.start();
         this.logScanHelper.start();
         this.runRecordingScanHelper.start();
-        this.flowScanHelper.start();
     }
 
     /**

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

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.job.utils.CronExpression;
+import com.byit.model.Flow;
+import com.byit.model.Node;
+import com.byit.service.FlowService;
+import com.byit.service.NodeService;
+import com.byit.service.RunNodeServer;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.text.ParseException;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 工作流操作  扫描待执行的非内嵌工作流加入运行实例并拆分节点至待执行任务表
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class FlowThreadRunHelper extends BaseThreadRunHelper {
+    private final Long PRE_TEST_TIME = System.currentTimeMillis()+ TimeUnit.HOURS.toMillis(30);
+    private static final String LOCK_NAME = "flow_lock";
+    private final DataSource dataSource;
+
+    private final FlowService flowService;
+    private final NodeService nodeService;
+    private final RunNodeServer runNodeServer;
+
+    public FlowThreadRunHelper(DataSource dataSource, FlowService flowService, NodeService nodeService, RunNodeServer runNodeServer) {
+        this.dataSource = dataSource;
+        this.flowService = flowService;
+        this.nodeService = nodeService;
+        this.runNodeServer = runNodeServer;
+    }
+
+    @Override
+    public boolean start() {
+        boolean isSleep = false;
+        //这个查询时有一个条件是 剩余次数不等于0也就是说   等于0的就查询不出来
+        List<Flow> halfAnHourFlow = flowService.findHalfAnHourFlow(PRE_TEST_TIME);
+        if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
+            for(Flow flow : halfAnHourFlow ){
+                log.debug("-----------------【工作流{}的执行次数大于0，放行】-------------------------",flow.getFlowName());
+                //String versionName = flow.getVersionName()
+                Integer flowId = flow.getFlowId();
+                //根据工作流查询工作流下所有的节点
+                List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
+                if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
+                    //保存到运行记录表和任务表
+                    runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
+                    if (flow.getRemainingCount()>0) {
+                        flow.setRemainingCount(flow.getRemainingCount()-1);
+                    }
+                    //获取cron表达式
+                    String flowCron = flow.getFlowCron();
+                    //设置下一周期的时间
+                    Date nextValidTime = null;
+                    try {
+                        nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
+                    } catch (ParseException e) {
+                        e.printStackTrace();
+                    }
+                    flow.setTriggerNextTime(nextValidTime!=null?nextValidTime.getTime():999999999999L);
+                    flowService.updateByIdSelective(flow);
+                }
+            }
+        }else{
+            isSleep = true;
+        }
+        return isSleep;
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+}