周期性调度功能添加

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

@@ -32,6 +32,11 @@ public class RunNodeServiceImpl implements RunNodeServer {
         this.jobTaskService = jobTaskService;
     }
 
+    /**
+     * 保存到运行记录一份  将节点保存到job_task表
+     * @param flow
+     * @param nodes
+     */
     @Override
     public void saveRunRecAndTask(Flow flow, List<Node> nodes) {
         log.info("---------saveRunRecAndTask start------【保存工作流：{}和节点：{}】-----------------------",flow,nodes);

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

 package com.byit.thread;
 
 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.*;
@@ -11,6 +12,8 @@ import javax.sql.DataSource;
 import java.sql.Connection;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.text.ParseException;
+import java.util.Date;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -64,23 +67,31 @@ public class FlowScanHelper {
                     conn.setAutoCommit(false);
                     preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'flow_lock' FOR UPDATE ");
                     preparedStatement.execute();
+                    //这个查询时有一个条件是 剩余次数不等于0也就是说   等于0的就查询不出来
                     List<Flow> halfAnHourFlow = flowService.findHalfAnHourFlow(PRE_TEST_TIME);
                     if (CollectionUtil.isNotEmpty(halfAnHourFlow)) {
-                        halfAnHourFlow.forEach(flow -> {
+                        for(Flow flow : halfAnHourFlow ){
                             log.debug("-----------------【工作流{}的执行次数大于0，放行】-------------------------",flow.getFlowName());
-                            String versionName = flow.getVersionName();
+                            //String versionName = flow.getVersionName()
                             Integer flowId = flow.getFlowId();
                             //根据工作流查询工作流下所有的节点
                             List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
                             if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
                                 //保存到运行记录表和任务表
                                 runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
-                                flow.setRemainingCount(flow.getRemainingCount()-1);
+                                if (flow.getRemainingCount()>0) {
+                                    flow.setRemainingCount(flow.getRemainingCount()-1);
+                                }
+                                //获取cron表达式
+                                String flowCron = flow.getFlowCron();
+                                //设置下一周期的时间
+                                Date nextValidTime = new CronExpression(flowCron).getNextValidTimeAfter(new Date(flow.getTriggerNextTime()));
+                                flow.setTriggerNextTime(nextValidTime.getTime());
                                 flowService.updateByIdSelective(flow);
                             }
 
 
-                        });
+                        }
                     }else{
                         isSleep = true;
                     }

demo-client/byit-demo-client/src/main/java/com/byit/job/TestAddFlow2.java

@@ -29,7 +29,7 @@ public class TestAddFlow2 {
                 .flowCron("0 0/1 * * * ? *")
                 .flowTimeout(TimeUnit.MINUTES.toMillis(30))
                 .priority("2")
-                .repeatCount(1)
+                .repeatCount(2)
                 .scheduleFollow("1")
                 .build();