【日志扫描】完成虚拟节点的判定完结和节点通知执行记录表完成

byit-myth-admin/src/main/resources/logback-spring.xml

@@ -62,7 +62,6 @@
     </appender>
 
     <logger name="com.byit.thread" level="debug" additivity="false">
-        <appender-ref ref="file-myth-job" />
         <appender-ref ref="console"/>
     </logger>
 

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

 package com.byit.conf;
 
-import com.byit.thread.JobFlowThread;
 import com.byit.thread.JobScheduleHelper;
+import com.byit.thread.LogScanHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
@@ -17,10 +17,15 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 public class MythJobScheduler implements InitializingBean, DisposableBean {
+    private final JobScheduleHelper jobScheduleHelper;
+    private final LogScanHelper logScanHelper;
+
     @Autowired
-    private JobScheduleHelper jobScheduleHelper;
-    @Autowired
-    private JobFlowThread jobFlowThread;
+    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper) {
+        this.jobScheduleHelper = jobScheduleHelper;
+        this.logScanHelper = logScanHelper;
+    }
+
     /**
      * 销毁方法
      * @throws Exception
@@ -38,8 +43,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
     public void afterPropertiesSet() throws Exception {
         //启用扫描线程
         jobScheduleHelper.start();
-
-        jobFlowThread.start();
+        logScanHelper.start();
 
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java

 package com.byit.mapper;
 
+import com.byit.model.JobTaskRunLog;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 import org.springframework.stereotype.Repository;
 
+import java.util.List;
+
 /**
  * 对日志表的操作
  * @author huangfu
@@ -10,6 +13,22 @@ import org.springframework.stereotype.Repository;
 @Repository
 public interface JobTaskRunLogMapper {
     /**
+     * 查询已经结束或者失败的节点
+     * @return
+     */
+    List<JobTaskRunLog> findJobTaskRunLogEndOrFailureNode();
+    /**
+     * 查询为结束的虚拟节点
+     * @return
+     */
+    List<JobTaskRunLog> findNotEndVirtualNode();
+    /**
+     * 根据ID查询一条日志
+     * @param logId
+     * @return
+     */
+    JobTaskRunLogWithBLOBs findJobTaskRunLogByLogId(Integer logId);
+    /**
      * 根据id删除一个数据
      * @param logId
      * @return
@@ -24,17 +43,17 @@ public interface JobTaskRunLogMapper {
     int saveJobTaskRunLog(JobTaskRunLogWithBLOBs record);
 
     /**
-     * 根据ID查询一条日志
-     * @param logId
+     * 修改一条日志
+     * @param record
      * @return
      */
-    JobTaskRunLogWithBLOBs findJobTaskRunLogByLogId(Integer logId);
+    int updateJobTaskRunLogWithBLOBs(JobTaskRunLogWithBLOBs record);
 
     /**
      * 修改一条日志
      * @param record
      * @return
      */
-    int updateJobTaskRunLog(JobTaskRunLogWithBLOBs record);
+    int updateJobTaskRunLog(JobTaskRunLog record);
 
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java

 package com.byit.mapper;
 
 import com.byit.model.RunRecording;
+import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
 
 /**
- * @author 删除数据
+ * 运行记录表
+ * @author huangfu
  */
+@Repository
 public interface RunRecordingMapper {
     /**
+     * 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
+     * @param flowId
+     * @param runId
+     * @return
+     */
+    RunRecording findRunRecordingByFlowIdAndRunId(@Param("flowId") Integer flowId, @Param("runId") String runId);
+    /**
      * 根据ID查询
      * @param recordingId
      * @return
@@ -25,6 +36,13 @@ public interface RunRecordingMapper {
      * @return
      */
     int updateRunRecordingById(RunRecording record);
+
+    /**
+     * 修改数据  根据工作流id和运行标识
+     * @param record
+     * @return
+     */
+    int updateRunRecordingByFlowIdAndRunId(RunRecording record);
     /**
      * 根据ID删除
      * @param recordingId

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java

@@ -144,11 +144,16 @@ public class JobTaskRunLog implements Serializable {
     @ApiModelProperty("任务类型")
     private String jobType;
     /**
-     * 任务类型
+     * 是否已完成告警(0未完成  1完成)
      */
-    @ApiModelProperty("任务类型")
+    @ApiModelProperty("是否已完成告警(0未完成  1完成)")
     private String alertEnd;
     /**
+     * 运行标识
+     */
+    @ApiModelProperty("运行标识")
+    private String runId;
+    /**
      */
     private static final long serialVersionUID = 1L;
 }
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/RunRecording.java

@@ -72,7 +72,7 @@ public class RunRecording implements Serializable {
      * 工作流的版本名称
      */
     @ApiModelProperty("工作流的版本名称")
-    private Integer flowVersionName;
+    private String flowVersionName;
 
     /**
      * 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)

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

@@ -3,6 +3,8 @@ package com.byit.service;
 import com.byit.model.JobTaskRunLog;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 
+import java.util.List;
+
 /**
  * @program: byit-myth-job->JobTaskRunLogService
  * @description: 日志业务表
@@ -11,6 +13,16 @@ import com.byit.model.JobTaskRunLogWithBLOBs;
  **/
 public interface JobTaskRunLogService {
     /**
+     * 查询已经结束或者失败的节点
+     * @return
+     */
+    List<JobTaskRunLog> findJobTaskRunLogEndOrFailureNode();
+    /**
+     * 查询为结束的虚拟节点
+     * @return
+     */
+    List<JobTaskRunLog> findNotEndVirtualNode();
+    /**
      * 查询一条数据  根据ID
      * @param id
      * @return
@@ -29,5 +41,11 @@ public interface JobTaskRunLogService {
      * @param jobTaskRunLogWithBLOBs
      * @return
      */
-    int updateJobTaskRunLog(JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs);
+    int updateJobTaskRunLogWithBLOBs(JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs);
+    /**
+     * 修改数据
+     * @param jobTaskRunLog
+     * @return
+     */
+    int updateJobTaskRunLog(JobTaskRunLog jobTaskRunLog);
 }

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

+package com.byit.service;
+
+import com.byit.model.RunRecording;
+import org.apache.ibatis.annotations.Param;
+
+/**
+ * 运行记录表
+ * @author huangfu
+ */
+public interface RunRecordingService {
+    /**
+     * 这个方法是会根据任务流的id和运行标识找到唯一对应的一个任务流，这个任务流就是一个虚拟节点
+     * @param flowId
+     * @param runId
+     * @return
+     */
+    RunRecording findRunRecordingByFlowIdAndRunId(Integer flowId, String runId);
+    /**
+     * 根据ID查询
+     * @param recordingId
+     * @return
+     */
+    RunRecording findRunRecordingById(Integer recordingId);
+    /**
+     * 保存数据
+     * @param record
+     * @return
+     */
+    int saveRunRecording(RunRecording record);
+
+    /**
+     * 修改数据
+     * @param record
+     * @return
+     */
+    int updateRunRecordingById(RunRecording record);
+    /**
+     * 修改数据  根据工作流id和运行标识
+     * @param record
+     * @return
+     */
+    int updateRunRecordingByFlowIdAndRunId(RunRecording record);
+
+    /**
+     * 根据ID删除
+     * @param recordingId
+     * @return
+     */
+    int deleteById(Integer recordingId);
+}
\ No newline at end of file

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

 package com.byit.service.impl;
 
 import com.byit.mapper.JobTaskRunLogMapper;
+import com.byit.model.JobTaskRunLog;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.service.JobTaskRunLogService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -8,6 +9,8 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.List;
+
 /**
  * @program: byit-myth-job->JobTaskRunLogServiceImpl
  * @description: 日志业务表
@@ -26,6 +29,16 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
     }
 
     @Override
+    public List<JobTaskRunLog> findJobTaskRunLogEndOrFailureNode() {
+        return jobTaskRunLogMapper.findJobTaskRunLogEndOrFailureNode();
+    }
+
+    @Override
+    public List<JobTaskRunLog> findNotEndVirtualNode() {
+        return jobTaskRunLogMapper.findNotEndVirtualNode();
+    }
+
+    @Override
     @Transactional(rollbackFor = Exception.class,propagation = Propagation.SUPPORTS)
     public JobTaskRunLogWithBLOBs findJobTaskRunLogById(Integer id) {
         return jobTaskRunLogMapper.findJobTaskRunLogByLogId(id);
@@ -37,7 +50,12 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
     }
 
     @Override
-    public int updateJobTaskRunLog(JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs) {
-        return jobTaskRunLogMapper.updateJobTaskRunLog(jobTaskRunLogWithBLOBs);
+    public int updateJobTaskRunLogWithBLOBs(JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs) {
+        return jobTaskRunLogMapper.updateJobTaskRunLogWithBLOBs(jobTaskRunLogWithBLOBs);
+    }
+
+    @Override
+    public int updateJobTaskRunLog(JobTaskRunLog jobTaskRunLog) {
+        return jobTaskRunLogMapper.updateJobTaskRunLog(jobTaskRunLog);
     }
 }

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

+package com.byit.service.impl;
+
+import com.byit.mapper.RunRecordingMapper;
+import com.byit.model.RunRecording;
+import com.byit.service.RunRecordingService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @program: byit-myth-job->RunRecordingServiceImpl
+ * @description: 对应的任务流运行记录操作
+ * @author: huangfu
+ * @date: 2019/12/26 15:08
+ **/
+@Service
+public class RunRecordingServiceImpl implements RunRecordingService {
+    private final RunRecordingMapper runRecordingMapper;
+
+    @Autowired
+    public RunRecordingServiceImpl(RunRecordingMapper runRecordingMapper) {
+        this.runRecordingMapper = runRecordingMapper;
+    }
+
+    @Override
+    public RunRecording findRunRecordingByFlowIdAndRunId(Integer flowId, String runId) {
+        return runRecordingMapper.findRunRecordingByFlowIdAndRunId(flowId,runId);
+    }
+
+    @Override
+    public RunRecording findRunRecordingById(Integer recordingId) {
+        return runRecordingMapper.findRunRecordingById(recordingId);
+    }
+
+    @Override
+    public int saveRunRecording(RunRecording record) {
+        return runRecordingMapper.saveRunRecording(record);
+    }
+
+    @Override
+    public int updateRunRecordingById(RunRecording record) {
+        return runRecordingMapper.updateRunRecordingById(record);
+    }
+
+    @Override
+    public int updateRunRecordingByFlowIdAndRunId(RunRecording record) {
+        return runRecordingMapper.updateRunRecordingByFlowIdAndRunId(record);
+    }
+
+    @Override
+    public int deleteById(Integer recordingId) {
+        return runRecordingMapper.deleteById(recordingId);
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java

@@ -78,6 +78,6 @@ public class JavaBeanJobTask implements TimerTask {
         jobTaskRunLog.setTriggerTime(new Date());
         jobTaskRunLog.setTriggerCode(dispatchResponseDto.getCode());
         jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
-        jobTaskRunLogService.updateJobTaskRunLog(jobTaskRunLog);
+        jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
     }
 }

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

-package com.byit.thread;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-import javax.sql.DataSource;
-
-/**
- * @program: byit-myth-job->JobFlowThread
- * @description: 扫描任务流表，关联版本表 获取任务流信息！
- *      将此工作流添加到执行记录表！将该工作流下的该版本所有节点添加到预读表！
- * @author: huangfu
- * @date: 2019/12/18 14:12
- **/
-@Slf4j
-@Component
-public class JobFlowThread {
-    private DataSource dataSource;
-
-    @Autowired
-    public void setDataSource(DataSource dataSource) {
-        this.dataSource = dataSource;
-    }
-
-    private Thread flowThread;
-
-    public void start(){
-        flowThread = new Thread(()->{
-
-        });
-    }
-
-}

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

@@ -37,19 +37,17 @@ public class JobScheduleHelper{
     private DataSource dataSource;
 
 
-    @Autowired
-    private JobTaskService jobTaskService;
-    @Autowired
-    private JobTaskScheduleService jobTaskScheduleService;
+    private final JobTaskService jobTaskService;
+    private final JobTaskScheduleService jobTaskScheduleService;
 
     /**
      * 读取任务节点的预读
      */
-    public static final long PRE_READ_MS = 7000;
+    private static final long PRE_READ_MS = 7000;
     /**
      * 读取任务排期表的预读
      */
-    public static final long SCHEDULE_READ_MS=5000;
+    private static final long SCHEDULE_READ_MS=5000;
     /**
      * 任务节点线程
      */
@@ -68,6 +66,12 @@ public class JobScheduleHelper{
      */
     private volatile boolean scheduleThreadToStop = false;
 
+    @Autowired
+    public JobScheduleHelper(JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService) {
+        this.jobTaskScheduleService = jobTaskScheduleService;
+        this.jobTaskService = jobTaskService;
+    }
+
     /**
      * 启动两条线程
      */
@@ -98,7 +102,7 @@ public class JobScheduleHelper{
                     //修改为不自动提交
                     conn.setAutoCommit(false);
                     //先添加扫描任务节点的行锁
-                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'scanning_job_read_ahead' FOR UPDATE ");
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_lock' FOR UPDATE ");
                     preparedStatement.execute();
                     //行锁已经加上  后续处理
                     long nowTime = System.currentTimeMillis();
@@ -232,7 +236,7 @@ public class JobScheduleHelper{
                     //修改为不自动提交
                     conn.setAutoCommit(false);
                     //先添加扫描任务节点的行锁
-                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'scanning_job_flight_schedule' FOR UPDATE ");
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'job_task_schedule_lock' FOR UPDATE ");
                     preparedStatement.execute();
                     //行锁已经加上  后续处理
                     long nowTime = System.currentTimeMillis();
@@ -325,9 +329,8 @@ public class JobScheduleHelper{
     private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
         JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
 
+        jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
         jobTaskRunLog.setFlowName(mythJobTaskSchedule.getVersionName());
-        //版本id需要查验
-        //还需要携带版本的名字
         jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
         jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
         jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());

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

@@ -31,6 +31,6 @@ public class LogCallbackThread implements Runnable {
         jobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
         jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
         jobTaskRunLog.setAlertEnd("0");
-        mythJobTaskRunLogService.updateJobTaskRunLog(jobTaskRunLog);
+        mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
     }
 }

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

+package com.byit.thread;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.model.JobTaskRunLog;
+import com.byit.model.RunRecording;
+import com.byit.service.JobTaskRunLogService;
+import com.byit.service.RunRecordingService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * @program: byit-myth-job->LogScanHelper
+ * @description: 对日志表的扫描
+ * @author: huangfu
+ * @date: 2019/12/26 10:34
+ **/
+@Component
+@Slf4j
+public class LogScanHelper {
+    private DataSource dataSource;
+    private final JobTaskRunLogService jobTaskRunLogService;
+    private final RunRecordingService runRecordingService;
+    @Autowired
+    public LogScanHelper(JobTaskRunLogService jobTaskRunLogService, RunRecordingService runRecordingService) {
+        this.jobTaskRunLogService = jobTaskRunLogService;
+        this.runRecordingService = runRecordingService;
+    }
+
+    @Autowired
+    public void setDataSource(DataSource dataSource) {
+        this.dataSource = dataSource;
+    }
+
+
+    private final String FAILURE_CODE = "100500";
+    private final String SUCCESS_CODE = "100200";
+    /**
+     * 扫描虚节点的线程是否停止
+     */
+    private volatile boolean virtualNodeScanIsStop = false;
+    /**
+     * 扫描未完成告警的节点的线程是否停止
+     */
+    private volatile boolean notAlarmedNodeScanIsStop = false;
+    /**
+     * 扫描虚节点的线程定义
+     */
+    private Thread virtualNodeScanThread = null;
+    /**
+     * 扫描未完成告警的节点的线程
+     */
+    private Thread notAlarmedNodeScanThread = null;
+
+    public void start(){
+        virtualNodeScanMethod();
+        notAlarmedNodeScanMethod();
+    }
+
+    /**
+     * 虚节点扫描
+     */
+    private void virtualNodeScanMethod(){
+        //扫描虚节点线程
+        virtualNodeScanThread = new Thread(() ->{
+            dateAligned(5000);
+            while (!virtualNodeScanIsStop){
+                Connection conn = null;
+                Boolean connAutoCommit = null;
+                PreparedStatement preparedStatement = null;
+                try{
+                    conn = dataSource.getConnection();
+                    connAutoCommit = conn.getAutoCommit();
+                    conn.setAutoCommit(false);
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_virtual_node_lock' FOR UPDATE ");
+                    preparedStatement.execute();
+                    //查询日志表没有完结的虚拟节点
+                    List<JobTaskRunLog> notEndVirtualNodes = jobTaskRunLogService.findNotEndVirtualNode( );
+                    if(CollectionUtil.isNotEmpty(notEndVirtualNodes)){
+                        notEndVirtualNodes.forEach(notEndVirtualNode ->{
+                            log.debug("------------【开始查询虚拟节点的执行情况】----------------");
+                            //根据运行标识和工作流id查询运行日志
+                            RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(notEndVirtualNode.getFlowId( ), notEndVirtualNode.getRunId( ));
+                            //判断当前的工作流是否已经完结
+                            if(runRecordingByFlowIdAndRunId!=null && "4".equals(runRecordingByFlowIdAndRunId.getFlowStatus())){
+                                log.debug("------------【查询到有已经完成的虚拟节点修改日志】----------------");
+                                notEndVirtualNode.setRunCode(runRecordingByFlowIdAndRunId.getFlowRunResult());
+                                //修改日志信息 改为成功或者失败
+                                jobTaskRunLogService.updateJobTaskRunLog(notEndVirtualNode);
+                            }else{
+                                dateAligned(1000);
+                            }
+                        });
+                    }else{
+                        dateAligned(20000);
+                    }
+                }catch (Exception e){
+                    e.printStackTrace();
+                }finally {
+                    //释放资源
+                    freedResource(conn,preparedStatement,connAutoCommit);
+                }
+            }
+
+        });
+        virtualNodeScanThread.setName("myth-job#【LogScanHelper】#virtualNodeScanThread");
+        virtualNodeScanThread.setDaemon(true);
+        virtualNodeScanThread.start();
+    }
+
+    /**
+     * 未告警的节点扫描
+     */
+    private void notAlarmedNodeScanMethod(){
+        notAlarmedNodeScanThread = new Thread(() ->{
+            while (!notAlarmedNodeScanIsStop){
+                dateAligned(5000);
+                Connection conn = null;
+                Boolean connAutoCommit = null;
+                PreparedStatement preparedStatement = null;
+                try {
+                    conn = dataSource.getConnection( );
+                    connAutoCommit = conn.getAutoCommit( );
+                    conn.setAutoCommit(false);
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'log_node_callback_lock' FOR UPDATE ");
+                    preparedStatement.execute();
+                    //查询的是失败的或者是已经结束的节点（完成的）
+                    List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
+                    if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
+                        log.debug("-------------【查询到有完结而且未告警的节点】-------------");
+                        Map<String, List<JobTaskRunLog>> jobLogMap = jobTaskRunLogEndOrFailureNode.stream( ).collect(Collectors.groupingBy(JobTaskRunLog::getRunCode));
+                        //假设成功的code码是000200    失败的是000500
+                        List<JobTaskRunLog> failureNodes = jobLogMap.get(FAILURE_CODE);
+                        //成功的
+                        List<JobTaskRunLog> successNodes = jobLogMap.get(SUCCESS_CODE);
+                        //遍历成功的节点  修改执行记录表
+                        if(CollectionUtil.isNotEmpty(successNodes)){
+                            successNodes.forEach(successNode ->{
+                                String runId = successNode.getRunId( );
+                                Integer flowId = successNode.getFlowId( );
+                                runRecordingService.updateRunRecordingByFlowIdAndRunId(RunRecording.builder().runId(runId).flowId(flowId).flowRunResult("1").flowStatus("4").build());
+                            });
+                        }else{
+                            dateAligned(1000);
+                        }
+                    }else{
+                        dateAligned(20000);
+                    }
+
+                }catch (Exception e){
+                    e.printStackTrace();
+                }finally {
+                    freedResource(conn,preparedStatement,connAutoCommit);
+                }
+            }
+        });
+        notAlarmedNodeScanThread.setName("myth-job#【LogScanHelper】#notAlarmedNodeScanThread");
+        notAlarmedNodeScanThread.setDaemon(true);
+        notAlarmedNodeScanThread.start();
+    }
+
+    /**
+     * 释放资源
+     * @param conn 数据库链接
+     * @param preparedStatement 执行器
+     * @param connAutoCommit 原来的提交状态
+     */
+    private void freedResource(Connection conn,PreparedStatement preparedStatement,boolean connAutoCommit){
+        if(conn != null){
+            try {
+                conn.commit();
+            } catch (SQLException e) {
+                if(!virtualNodeScanIsStop){
+                    log.error("--------------------【提交行锁出错】---------------------");
+                }
+            }
+        }
+
+        try {
+            assert conn != null;
+            conn.setAutoCommit(connAutoCommit);
+        } catch (SQLException e) {
+            if(!virtualNodeScanIsStop){
+                log.error("--------------------【恢复自动提交出错】---------------------");
+            }
+        }
+
+        try {
+            conn.close();
+        } catch (SQLException e) {
+            if(!virtualNodeScanIsStop){
+                log.error("--------------------【关闭链接出错】---------------------");
+            }
+        }
+
+        if(preparedStatement != null){
+            try {
+                preparedStatement.close();
+            } catch (SQLException e) {
+                if(!virtualNodeScanIsStop){
+                    log.error("--------------------【关闭执行器出错】---------------------");
+                }
+            }
+        }
+    }
+
+    /**
+     * 对齐时钟。整秒运行
+     */
+    private void dateAligned(long waitTime){
+        try {
+            TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace( );
+        }
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/util/SourceObj2TargetObjUtil.java

@@ -3,6 +3,8 @@ package com.byit.util;
 import com.byit.job.dto.PluginBeanJobInfo;
 import com.byit.model.JobTask;
 
+import java.util.UUID;
+
 /**
  * @program: byit-myth-job->SourceObj2TargetObjUtil
  * @description: 这是一个转换的工具类，定义两个类之间的相互转换
@@ -31,6 +33,7 @@ public class SourceObj2TargetObjUtil {
         jobTask.setTriggerTime(System.currentTimeMillis()+20000);
         jobTask.setJobType("JAVA");
         jobTask.setVersionName("V1");
+        jobTask.setRunId(UUID.randomUUID( ).toString());
         return jobTask;
     }
 }

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml

@@ -24,6 +24,7 @@
     <result column="node_id" jdbcType="INTEGER" property="nodeId" />
     <result column="alert_end" jdbcType="CHAR" property="alertEnd" />
     <result column="job_type" jdbcType="CHAR" property="jobType" />
+    <result column="run_id" jdbcType="VARCHAR" property="runId" />
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
     <result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
@@ -32,12 +33,23 @@
   <sql id="Base_Column_List">
     log_id, failed_remaining_count, version_name, flow_id, flow_name, job_group_id, handler_name, 
     node_name, is_virtual, run_code, run_params, start_time, run_type, trigger_code, 
-    trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end
+    trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end,run_id
   </sql>
   <sql id="Blob_Column_List">
     run_msg, trigger_msg
   </sql>
 
+  <select id="findJobTaskRunLogEndOrFailureNode" resultMap="BaseResultMap">
+
+    select  <include refid="Base_Column_List" /> from job_task_run_log
+    where (node_name='END' or run_code = '000500') and alert_end = '0'
+  </select>
+
+  <select id="findNotEndVirtualNode" resultMap="BaseResultMap">
+    select  <include refid="Base_Column_List" /> from job_task_run_log
+      where is_virtual='0' and (run_code is null	or run_code = '')
+  </select>
+
   <select id="findJobTaskRunLogByLogId" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
     select 
     <include refid="Base_Column_List" />
@@ -127,6 +139,9 @@
       <if test="triggerMsg != null">
         trigger_msg,
       </if>
+      <if test="runId != null">
+        run_id,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
       <if test="logId != null">
@@ -201,9 +216,12 @@
       <if test="triggerMsg != null">
         #{triggerMsg,jdbcType=LONGVARCHAR},
       </if>
+      <if test="runId != null">
+        run_id = #{runId,jdbcType=VARCHAR},
+      </if>
     </trim>
   </insert>
-  <update id="updateJobTaskRunLog" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
+  <update id="updateJobTaskRunLogWithBLOBs" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
     update job_task_run_log
     <set>
       <if test="failedRemainingCount != null">
@@ -275,6 +293,81 @@
       <if test="triggerMsg != null">
         trigger_msg = #{triggerMsg,jdbcType=LONGVARCHAR},
       </if>
+      <if test="runId != null">
+        run_id = #{runId,jdbcType=VARCHAR},
+      </if>
+    </set>
+    where log_id = #{logId,jdbcType=INTEGER}
+  </update>
+  <update id="updateJobTaskRunLog" parameterType="com.byit.model.JobTaskRunLog">
+    update job_task_run_log
+    <set>
+      <if test="failedRemainingCount != null">
+        failed_remaining_count = #{failedRemainingCount,jdbcType=INTEGER},
+      </if>
+      <if test="versionName != null">
+        version_name = #{versionName,jdbcType=VARCHAR},
+      </if>
+      <if test="flowId != null">
+        flow_id = #{flowId,jdbcType=INTEGER},
+      </if>
+      <if test="flowName != null">
+        flow_name = #{flowName,jdbcType=VARCHAR},
+      </if>
+      <if test="jobGroupId != null">
+        job_group_id = #{jobGroupId,jdbcType=INTEGER},
+      </if>
+      <if test="handlerName != null">
+        handler_name = #{handlerName,jdbcType=VARCHAR},
+      </if>
+      <if test="nodeName != null">
+        node_name = #{nodeName,jdbcType=VARCHAR},
+      </if>
+      <if test="isVirtual != null">
+        is_virtual = #{isVirtual,jdbcType=CHAR},
+      </if>
+      <if test="runCode != null">
+        run_code = #{runCode,jdbcType=VARCHAR},
+      </if>
+      <if test="runParams != null">
+        run_params = #{runParams,jdbcType=VARCHAR},
+      </if>
+      <if test="startTime != null">
+        start_time = #{startTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="runType != null">
+        run_type = #{runType,jdbcType=CHAR},
+      </if>
+      <if test="triggerCode != null">
+        trigger_code = #{triggerCode,jdbcType=VARCHAR},
+      </if>
+      <if test="triggerTime != null">
+        trigger_time = #{triggerTime,jdbcType=DATE},
+      </if>
+      <if test="jobGroupIp != null">
+        job_group_ip = #{jobGroupIp,jdbcType=VARCHAR},
+      </if>
+      <if test="mapFlowId != null">
+        map_flow_id = #{mapFlowId,jdbcType=INTEGER},
+      </if>
+      <if test="runCommand != null">
+        run_command = #{runCommand,jdbcType=VARCHAR},
+      </if>
+      <if test="endTime != null">
+        end_time = #{endTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="nodeId != null">
+        node_id = #{nodeId,jdbcType=INTEGER},
+      </if>
+      <if test="jobType != null">
+        job_type = #{jobType,jdbcType=VARCHAR},
+      </if>
+      <if test="alertEnd != null">
+        alert_end = #{alertEnd,jdbcType=VARCHAR},
+      </if>
+      <if test="runId != null">
+        run_id = #{runId,jdbcType=VARCHAR},
+      </if>
     </set>
     where log_id = #{logId,jdbcType=INTEGER}
   </update>

byit-myth-core/myth-admin-core/src/main/resources/mapper/RunRecordingMapper.xml

@@ -2,7 +2,6 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="com.byit.mapper.RunRecordingMapper">
   <resultMap id="BaseResultMap" type="com.byit.model.RunRecording">
-    <!-- generated @mbg.generated date: 2019-12-25 -->
     <id column="recording_id" jdbcType="INTEGER" property="recordingId" />
     <result column="run_id" jdbcType="VARCHAR" property="runId" />
     <result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" />
@@ -11,7 +10,7 @@
     <result column="flow_run_result" jdbcType="CHAR" property="flowRunResult" />
     <result column="flow_status" jdbcType="CHAR" property="flowStatus" />
     <result column="flow_timeout" jdbcType="BIGINT" property="flowTimeout" />
-    <result column="flow_version_name" jdbcType="INTEGER" property="flowVersionName" />
+    <result column="flow_version_name" jdbcType="VARCHAR" property="flowVersionName" />
     <result column="alarml_action" jdbcType="CHAR" property="alarmlAction" />
     <result column="priority" jdbcType="CHAR" property="priority" />
     <result column="trigger_time" jdbcType="BIGINT" property="triggerTime" />
@@ -23,14 +22,20 @@
     <result column="is_inner" jdbcType="CHAR" property="isInner" />
   </resultMap>
   <sql id="Base_Column_List">
-    <!-- generated @mbg.generated date: 2019-12-25 -->
     recording_id, run_id, alarm_email, dispatch_ip, flow_name, flow_run_result, flow_status, 
     flow_timeout, flow_version_name, alarml_action, priority, trigger_time, principal, 
     flow_id, start_time, end_time, is_alarm,is_inner
   </sql>
+
+  <select id="findRunRecordingByFlowIdAndRunId" resultMap="BaseResultMap" >
+    select
+    <include refid="Base_Column_List" />
+    from run_recording
+    where run_id = #{runId,jdbcType=VARCHAR} and flow_id = #{flowId,jdbcType=INTEGER}
+  </select>
+
   <select id="findRunRecordingById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
-    <!-- generated @mbg.generated date: 2019-12-25 -->
-    select 
+    select
     <include refid="Base_Column_List" />
     from run_recording
     where recording_id = #{recordingId,jdbcType=INTEGER}
@@ -125,7 +130,7 @@
         #{flowTimeout,jdbcType=BIGINT},
       </if>
       <if test="flowVersionName != null">
-        #{flowVersionName,jdbcType=INTEGER},
+        #{flowVersionName,jdbcType=VARCHAR},
       </if>
       <if test="alarmlAction != null">
         #{alarmlAction,jdbcType=CHAR},
@@ -182,7 +187,7 @@
         flow_timeout = #{flowTimeout,jdbcType=BIGINT},
       </if>
       <if test="flowVersionName != null">
-        flow_version_name = #{flowVersionName,jdbcType=INTEGER},
+        flow_version_name = #{flowVersionName,jdbcType=VARCHAR},
       </if>
       <if test="alarmlAction != null">
         alarml_action = #{alarmlAction,jdbcType=CHAR},
@@ -214,4 +219,62 @@
     </set>
     where recording_id = #{recordingId,jdbcType=INTEGER}
   </update>
+  <update id="updateRunRecordingByFlowIdAndRunId" parameterType="com.byit.model.RunRecording">
+    <!-- generated @mbg.generated date: 2019-12-25 -->
+    update run_recording
+    <set>
+      <if test="runId != null">
+        run_id = #{runId,jdbcType=VARCHAR},
+      </if>
+      <if test="alarmEmail != null">
+        alarm_email = #{alarmEmail,jdbcType=VARCHAR},
+      </if>
+      <if test="dispatchIp != null">
+        dispatch_ip = #{dispatchIp,jdbcType=VARCHAR},
+      </if>
+      <if test="flowName != null">
+        flow_name = #{flowName,jdbcType=VARCHAR},
+      </if>
+      <if test="flowRunResult != null">
+        flow_run_result = #{flowRunResult,jdbcType=CHAR},
+      </if>
+      <if test="flowStatus != null">
+        flow_status = #{flowStatus,jdbcType=CHAR},
+      </if>
+      <if test="flowTimeout != null">
+        flow_timeout = #{flowTimeout,jdbcType=BIGINT},
+      </if>
+      <if test="flowVersionName != null">
+        flow_version_name = #{flowVersionName,jdbcType=VARCHAR},
+      </if>
+      <if test="alarmlAction != null">
+        alarml_action = #{alarmlAction,jdbcType=CHAR},
+      </if>
+      <if test="priority != null">
+        priority = #{priority,jdbcType=CHAR},
+      </if>
+      <if test="triggerTime != null">
+        trigger_time = #{triggerTime,jdbcType=BIGINT},
+      </if>
+      <if test="principal != null">
+        principal = #{principal,jdbcType=VARCHAR},
+      </if>
+      <if test="flowId != null">
+        flow_id = #{flowId,jdbcType=INTEGER},
+      </if>
+      <if test="startTime != null">
+        start_time = #{startTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="endTime != null">
+        end_time = #{endTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="isAlarm != null">
+        is_alarm = #{isAlarm,jdbcType=CHAR},
+      </if>
+      <if test="isInner != null">
+        is_inner = #{isInner,jdbcType=CHAR},
+      </if>
+    </set>
+    where flow_id = #{flowId,jdbcType=INTEGER} and run_id = #{runId,jdbcType=VARCHAR}
+  </update>
 </mapper>
\ No newline at end of file

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

@@ -10,7 +10,7 @@ import com.byit.job.vo.ReturnResult;
  * @author: huangfu
  * @date: 2019/11/20 12:40
  **/
-@JobHandler("addJob1")
+@JobHandler("END")
 public class DemoJob1 extends BaseJobHandler {
     @Override
     public ReturnResult<String> execute(String s) throws Exception {

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

@@ -26,12 +26,12 @@ public class Mains {
         String requestPort="8080";
         String param="不延迟任务";
         String name="addJob";
-        String email = "huangfusuper163.com";
+        String email = "huangfusuper@163.com";
         String email1 = "huangfusuper@163.com";
         PluginBeanJobInfo pluginBeanJobInfo = new PluginBeanJobInfo(name,plServerUrl,mythCron,routingStrategy,blockingStrategy,callbackToken,gatewayToken,requestIP,requestPort,param,email,"");
         System.out.println(pluginBeanJobInfo);
         PluginBeanJobInfo javaBeanJobInfo1 = new PluginBeanJobInfo(name,plServerUrl,mythCron,routingStrategy,blockingStrategy,callbackToken,gatewayToken,requestIP,requestPort,param,email1,"");
-        javaBeanJobInfo1.setJobHandelName("addJob1");
+        javaBeanJobInfo1.setJobHandelName("END");
         javaBeanJobInfo1.setParam("延迟任务");
         JobUtils.addJob(javaBeanJobInfo1);
         JobUtils.addJob(pluginBeanJobInfo);