排期表运行实例扫描线程重构

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

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.enums.NodeTypeEnum;
+import com.byit.job.WorkRoulette;
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.model.JobTaskSchedule;
+import com.byit.service.JobTaskScheduleService;
+import com.byit.service.impl.JobTaskRunLogServiceImpl;
+import com.byit.task.JavaBeanJobTask;
+import com.byit.task.ScriptExecutorJobTask;
+import com.byit.thread.BaseThreadRunHelper;
+import com.byit.util.SpringUtil;
+import io.netty.util.TimerTask;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+
+/**
+ * 排期表操作
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
+    /**
+     * 读取任务排期表的预读
+     */
+    private static final long SCHEDULE_READ_MS=5000;
+    private static final String LOCK_NAME = "job_task_schedule_lock";
+    private final DataSource dataSource;
+
+    private final JobTaskScheduleService jobTaskScheduleService;
+
+    public ScheduleThreadRunHelper(DataSource dataSource, JobTaskScheduleService jobTaskScheduleService) {
+        this.dataSource = dataSource;
+        this.jobTaskScheduleService = jobTaskScheduleService;
+    }
+
+    @Override
+    public Long start() {
+        long nowTime = System.currentTimeMillis();
+        //查询所有符合条件的任务节点
+        List<JobTaskSchedule> jobTaskSchedules = jobTaskScheduleService.findJobTaskScheduleByTriggerNextTimeLessThanEqual(nowTime + SCHEDULE_READ_MS);
+        if(CollectionUtil.isNotEmpty(jobTaskSchedules)){
+            log.info("------排期表查询到有需要存在的节点--------");
+            //循环遍历添加任务
+            jobTaskSchedules.forEach(mythJobTaskSchedule ->{
+                //如果是重跑就有logId
+                Integer logId = mythJobTaskSchedule.getLogId();
+                if(logId == null){
+                    log.info("-------------发现节点{}不是重跑,执行日志节点数据初始化-----------",mythJobTaskSchedule);
+                    logId = saveLog(mythJobTaskSchedule);
+                    log.info("------------{}节点的日志保存成功,日志ID为{}---------------",mythJobTaskSchedule,logId);
+                }
+                mythJobTaskSchedule.setLogId(logId);
+                Long triggerTime = mythJobTaskSchedule.getTriggerTime();
+                TimerTask timerTask = null;
+                if (NodeTypeEnum.JAVA.getType().equals(mythJobTaskSchedule.getJobType())) {
+                    log.info("------节点{}，开始构建java执行器-------",mythJobTaskSchedule);
+                    //构建调度执行器
+                    timerTask = new JavaBeanJobTask(mythJobTaskSchedule);
+                }else if(NodeTypeEnum.SCRIPT.getType().equals(mythJobTaskSchedule.getJobType())){
+                    log.info("------节点{}，开始构建脚本执行器-------",mythJobTaskSchedule);
+                    //构建脚本调度执行器
+                    timerTask = new ScriptExecutorJobTask(mythJobTaskSchedule);
+                }
+                //TODO 有个坑 如果这个类型不存在的话  这个节点就不会被执行和删除  有没有办法能够强制必须有类型呢？
+                if(timerTask != null){
+                    log.info("------节点{}的执行器{}执行添加到任务调度轮的操作-------",mythJobTaskSchedule,timerTask);
+                    WorkRoulette.addJob(timerTask,triggerTime);
+                    jobTaskScheduleService.delete(mythJobTaskSchedule.getId());
+                }
+
+            });
+        }else{
+            return SCHEDULE_READ_MS;
+        }
+        return NOT_WAIT_TIME;
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+
+    private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
+        JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+
+        jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
+        jobTaskRunLog.setIsVirtual(mythJobTaskSchedule.getIsVirtual());
+        jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
+        jobTaskRunLog.setFlowName(mythJobTaskSchedule.getFlowName());
+        jobTaskRunLog.setNodeId(mythJobTaskSchedule.getNodeId());
+        jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
+        jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
+        jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
+        jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
+        JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
+        mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
+
+        return jobTaskRunLog.getLogId();
+    }
+}