【插件测试】脚本执行插件测试

.gitignore

@@ -2,4 +2,7 @@
 *.iml
 .idea
 
-target
+target
\ No newline at end of file
+
+module_*.xml
+byit-myth-job.xml
\ No newline at end of file

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

@@ -53,6 +53,9 @@ public class JobTask implements Serializable {
     @ApiModelProperty("工作流ID")
     private Integer flowId;
 
+    @ApiModelProperty("工作流的名称")
+    private String flowName;
+
     /**
      * 插件端请求调度中心的令牌
      */

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

 package com.byit.model;
 
-import com.fasterxml.jackson.annotation.JsonFormat;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
 import java.io.Serializable;
 import java.util.Date;
-import lombok.Data;
 
 /**
- *  
+ * @author huangfu
  */
 @ApiModel
 @Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class JobTaskRunLog implements Serializable {
     /**
      * 日志ID
@@ -159,6 +165,7 @@ public class JobTaskRunLog implements Serializable {
     private String reRunId;
 
     /**
+     *
      */
     private static final long serialVersionUID = 1L;
 }
\ No newline at end of file

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

@@ -3,13 +3,17 @@ package com.byit.model;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import java.io.Serializable;
-import lombok.Data;
+
+import lombok.*;
 
 /**
  *  
  */
+@EqualsAndHashCode(callSuper = true)
 @ApiModel
 @Data
+@AllArgsConstructor
+@NoArgsConstructor
 public class JobTaskRunLogWithBLOBs extends JobTaskRunLog implements Serializable {
     /**
      * 运行结果信息

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

@@ -16,6 +16,9 @@ public class JobTaskSchedule implements Serializable {
     @ApiModelProperty("")
     private Integer id;
 
+    @ApiModelProperty("工作流的名称")
+    private String flowName;
+
     /**
      * 当前版本节点主键
      */

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

@@ -67,7 +67,7 @@ public class EmailAlarmVo implements Serializable {
      */
     @ApiModelProperty("告警邮箱")
     @NotBank(errorMessage = "告警邮箱不能为空，多个联系人可用,分割")
-    @EmailValidation
+    @EmailValidation(isEmails = true)
     private String alarmEmail;
 
     /**

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

+package com.byit.service;
+
+import com.byit.model.JobTask;
+import org.springframework.transaction.annotation.Transactional;
+
+/**
+ * @author 任务表和日志表的映射业务类，保证是一个事务+原子性
+ */
+public interface TaskAndLogServer {
+    /**
+     * 添加失败日志，删除任务表的任务
+     * @param jobTask
+     * @param runCode
+     */
+    void addRunLogAndRemoveTask(JobTask jobTask,String runCode);
+}

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

+package com.byit.service.impl;
+
+import com.byit.model.JobTask;
+import com.byit.model.JobTaskRunLogWithBLOBs;
+import com.byit.service.JobTaskRunLogService;
+import com.byit.service.JobTaskService;
+import com.byit.service.TaskAndLogServer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.Date;
+
+/**
+ * @author huangfu
+ */
+@Transactional(rollbackFor = Exception.class)
+@Service
+@Slf4j
+public class TaskAndLogServerImpl implements TaskAndLogServer {
+    private final JobTaskService jobTaskService;
+    private final JobTaskRunLogService jobTaskRunLogService;
+
+    public TaskAndLogServerImpl(JobTaskService jobTaskService, JobTaskRunLogService jobTaskRunLogService) {
+        this.jobTaskService = jobTaskService;
+        this.jobTaskRunLogService = jobTaskRunLogService;
+    }
+
+    @Override
+    public void addRunLogAndRemoveTask(JobTask jobTask,String runCode) {
+        Date thisDate = new Date();
+        //删除任务节点
+        jobTaskService.removeMythJobTaskById(jobTask.getId());
+        //添加任务日志
+        JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+        jobTaskRunLog.setReRunId(jobTask.getRunId());
+        jobTaskRunLog.setNodeId(jobTask.getNodeId());
+        jobTaskRunLog.setNodeName(jobTask.getNodeName());
+        jobTaskRunLog.setJobType(jobTask.getJobType());
+        jobTaskRunLog.setFlowId(jobTask.getFlowId());
+        jobTaskRunLog.setFailedRemainingCount(jobTask.getFailedRetryCount());
+        jobTaskRunLog.setVersionName(jobTask.getVersionName());
+        jobTaskRunLog.setFlowName(jobTask.getFlowName());
+        jobTaskRunLog.setHandlerName(jobTask.getHandlerName());
+        jobTaskRunLog.setIsVirtual(jobTask.getIsVirtual());
+        jobTaskRunLog.setMapFlowId(jobTask.getMapFlowId());
+        jobTaskRunLog.setRunCode(runCode);
+        jobTaskRunLog.setRunMsg("上级节点执行失败");
+        jobTaskRunLog.setRunParams(jobTask.getRunParam());
+        jobTaskRunLog.setRunCommand(jobTask.getRunCommand());
+        jobTaskRunLog.setRunType(jobTask.getJobType());
+        jobTaskRunLog.setTriggerCode("2");
+        jobTaskRunLog.setTriggerMsg("未执行调度");
+        jobTaskRunLog.setTriggerTime(thisDate);
+        jobTaskRunLog.setStartTime(thisDate);
+        jobTaskRunLog.setEndTime(thisDate);
+        jobTaskRunLog.setAlertEnd("1");
+        //添加日志节点
+        jobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
+    }
+}

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

@@ -26,19 +26,15 @@ public class FlowScanHelper {
     private final FlowService flowService;
     private final DataSource dataSource;
     private final NodeService nodeService;
-    private final JobTaskService jobTaskService;
-    private final RunRecordingService runRecordingService;
     private final RunNodeServer runNodeServer;
 
     private Thread flowThread;
     private volatile boolean flowThreadIsStop = false;
 
-    public FlowScanHelper(FlowService flowService, DataSource dataSource, NodeService nodeService, JobTaskService jobTaskService, RunRecordingService runRecordingService, RunNodeServer runNodeServer) {
+    public FlowScanHelper(FlowService flowService, DataSource dataSource, NodeService nodeService, RunNodeServer runNodeServer) {
         this.flowService = flowService;
         this.dataSource = dataSource;
         this.nodeService = nodeService;
-        this.jobTaskService = jobTaskService;
-        this.runRecordingService = runRecordingService;
         this.runNodeServer = runNodeServer;
     }
 
@@ -75,9 +71,12 @@ public class FlowScanHelper {
                             String versionName = flow.getVersionName();
                             Integer flowId = flow.getFlowId();
                             List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId, versionName);
-                            runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
-                            flow.setRemainingCount(flow.getRemainingCount()-1);
-                            flowService.updateByIdSelective(flow);
+                            if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
+                                runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
+                                flow.setRemainingCount(flow.getRemainingCount()-1);
+                                flowService.updateByIdSelective(flow);
+                            }
+
 
                         });
                     }else{
@@ -129,7 +128,7 @@ public class FlowScanHelper {
                 if(isSleep){
                     try {
                         log.info("------------------【未扫描到要执行的工作流】-------------------");
-                        TimeUnit.HOURS.sleep(10);
+                        TimeUnit.MINUTES.sleep(1);
                     } catch (InterruptedException e) {
                         log.warn("----------------------【扫描工作流的线程被关闭了】----------------------------");
                     }

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

@@ -6,10 +6,7 @@ import com.byit.model.JobTask;
 import com.byit.model.JobTaskRunLog;
 import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.model.JobTaskSchedule;
-import com.byit.service.JobTaskRunLogService;
-import com.byit.service.JobTaskScheduleService;
-import com.byit.service.JobTaskService;
-import com.byit.service.NodeDependencyService;
+import com.byit.service.*;
 import com.byit.service.impl.JobTaskRunLogServiceImpl;
 import com.byit.task.JavaBeanJobTask;
 import com.byit.util.SpringUtil;
@@ -44,6 +41,7 @@ public class JobScheduleHelper{
     private final JobTaskScheduleService jobTaskScheduleService;
     private final NodeDependencyService nodeDependencyService;
     private final JobTaskRunLogService jobTaskRunLogService;
+    private final TaskAndLogServer taskAndLogServer;
 
     /**
      * 读取任务节点的预读
@@ -72,11 +70,12 @@ public class JobScheduleHelper{
     private volatile boolean scheduleThreadToStop = false;
 
     @Autowired
-    public JobScheduleHelper(JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService) {
+    public JobScheduleHelper(JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer) {
         this.jobTaskScheduleService = jobTaskScheduleService;
         this.jobTaskService = jobTaskService;
         this.nodeDependencyService = nodeDependencyService;
         this.jobTaskRunLogService = jobTaskRunLogService;
+        this.taskAndLogServer = taskAndLogServer;
     }
 
     /**
@@ -127,12 +126,6 @@ public class JobScheduleHelper{
                     if(CollectionUtil.isNotEmpty(jobTasks)){
                         List<JobTaskSchedule> jobTaskSchedules = new ArrayList<JobTaskSchedule>(15);
                         jobTasks.forEach(jobTask -> {
-                            /**
-                             * 需要去检验当前任务的上级节点是否已经执行成功，没有执行，或者处于暂停状态则跳过该任务
-                             * 大概思路，根据任务流id，从任务流执行回溯表查询该任务流的所有节点，查看上级节点是否已经执行成功
-                             * //TODO  需要修改  判断父节点是否执行完毕  注意  父节点是一个集合
-                             */
-
                             //虚节点的状态
                             if("0".equals(jobTask.getIsVirtual())){
                                 //在日志表里面创建一条记录
@@ -158,6 +151,14 @@ public class JobScheduleHelper{
                                     for (JobTaskRunLog jobTaskRunLog :jobTaskRunLogList){
                                         //如果该节点不是成功的或者补批成功的  那么就跳过
                                         if(!("1".equals(jobTaskRunLog.getRunCode()) || "3".equals(jobTaskRunLog.getRunCode()))){
+                                            if(!("0".equals(jobTaskRunLog.getRunCode()))){
+                                                //记录状态码
+                                                String runCode = jobTaskRunLog.getRunCode();
+                                                log.info("-------------------【查询到有失败的节点，运行状态为：{}】-------------------",runCode);
+                                                //删除这个数据 并且添加到日志
+                                                taskAndLogServer.addRunLogAndRemoveTask(jobTask,runCode);
+                                            }
+
                                             return;
                                         }
                                     }
@@ -416,7 +417,7 @@ public class JobScheduleHelper{
 
         jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
         jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
-        jobTaskRunLog.setFlowName("1");
+        jobTaskRunLog.setFlowName(mythJobTaskSchedule.getFlowName());
         jobTaskRunLog.setNodeId(mythJobTaskSchedule.getNodeId());
         jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
         jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());

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

@@ -27,10 +27,12 @@ public class LogCallbackThread implements Runnable {
         JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
         JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
         jobTaskRunLog.setLogId(jobRunResultDto.getLogId());
-        jobTaskRunLog.setStartTime(jobRunResultDto.getEndTime());
+        jobTaskRunLog.setStartTime(jobRunResultDto.getStartTime());
+        jobTaskRunLog.setEndTime(jobRunResultDto.getEndTime());
         jobTaskRunLog.setRunCode(jobRunResultDto.getReturnResult().getCode());
         jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
         jobTaskRunLog.setAlertEnd("0");
         mythJobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
+
     }
 }

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

@@ -175,7 +175,7 @@ public class LogScanHelper {
                     List<JobTaskRunLog> jobTaskRunLogEndOrFailureNode = jobTaskRunLogService.findJobTaskRunLogEndOrFailureNode( );
                     if(CollectionUtil.isNotEmpty(jobTaskRunLogEndOrFailureNode)){
                         log.debug("-------------【查询到有完结而且未告警的节点】-------------");
-                        //遍历成功的节点  修改执行记录表
+                        //遍历结束的节点  修改执行记录表
                         jobTaskRunLogEndOrFailureNode.forEach(endNode ->{
                             String runId = endNode.getRunId( );
                             Integer flowId = endNode.getFlowId( );
@@ -185,7 +185,6 @@ public class LogScanHelper {
                             endNode.setAlertEnd("1");
                             jobTaskRunLogService.updateJobTaskRunLog(endNode);
                         });
-
                     }else{
                         dateAligned(20000);
                     }

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

@@ -9,6 +9,7 @@ import com.byit.model.RunRecording;
 import com.byit.service.EmailAlarmService;
 import com.byit.service.JobTaskRunLogService;
 import com.byit.service.RunRecordingService;
+import com.byit.util.TimeFormatUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -230,12 +231,12 @@ public class RunRecordingScanHelper {
 
         if (CollectionUtil.isNotEmpty(jobTaskRunLogs)) {
             jobTaskRunLogs.forEach(jobTaskRunLog -> {
-                long timeConsuming = TimeUnit.MILLISECONDS.toSeconds(jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime());
+                long timeConsuming = TimeUnit.MILLISECONDS.toMinutes(jobTaskRunLog.getEndTime().getTime() - jobTaskRunLog.getStartTime().getTime());
                 stringBuilder.append("<tr align='center'>")
                         .append(String.format("<td>%s</td>", jobTaskRunLog.getNodeName()))
                         .append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getStartTime(), DATE_FORMAT)))
                         .append(String.format("<td>%s</td>", DateUtil.format(jobTaskRunLog.getEndTime(), DATE_FORMAT)))
-                        .append(String.format("<td>%s</td>", timeConsuming + "分钟"))
+                        .append(String.format("<td>%s</td>", TimeFormatUtil.timeFormat(timeConsuming)))
                         .append(String.format("<td>%s</td>", "1".equals(jobTaskRunLog.getRunCode()) ? "成功"
                                 : "3".equals(jobTaskRunLog.getRunCode()) ? "补批成功"
                                 : "4".equals(jobTaskRunLog.getRunCode()) ? "补批失败" : "失败"))

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

@@ -30,6 +30,7 @@
     <result column="trigger_status" jdbcType="CHAR" property="triggerStatus" />
     <result column="version_name" jdbcType="VARCHAR" property="versionName" />
     <result column="run_command" jdbcType="VARCHAR" property="runCommand" />
+    <result column="flow_name" jdbcType="VARCHAR" property="flowName" />
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTask">
     <!-- generated @mbg.generated date: 2019-12-25 -->
@@ -41,7 +42,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual, 
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param, 
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name, 
-    run_command
+    run_command,flow_name
   </sql>
   <sql id="Blob_Column_List">
     run_source
@@ -154,6 +155,9 @@
       <if test="runSource != null">
         run_source,
       </if>
+      <if test="flowName != null">
+        flow_name,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
       <if test="id != null">
@@ -240,6 +244,9 @@
       <if test="runSource != null">
         #{runSource,jdbcType=LONGVARCHAR},
       </if>
+      <if test="flowName != null">
+        #{flowName,jdbcType=VARCHAR},
+      </if>
     </trim>
   </insert>
 
@@ -249,7 +256,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual,
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param,
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name,
-    run_command,run_source
+    run_command,run_source,flow_name
     ) values
     <foreach collection="jobTasks" item="jobTask" separator =",">
       (
@@ -260,7 +267,9 @@
       #{jobTask.failedRetryInterval,jdbcType=BIGINT},#{jobTask.routingStrategy,jdbcType=VARCHAR},#{jobTask.runId,jdbcType=VARCHAR},
       #{jobTask.runParam,jdbcType=VARCHAR},#{jobTask.runSourceDesc,jdbcType=VARCHAR},#{jobTask.scriptUrls,jdbcType=VARCHAR},
       #{jobTask.sourcePrincipal,jdbcType=VARCHAR},#{jobTask.triggerTime,jdbcType=BIGINT},#{jobTask.triggerStatus,jdbcType=CHAR},
-      #{jobTask.versionName,jdbcType=VARCHAR},#{jobTask.runCommand,jdbcType=VARCHAR},#{jobTask.runSource,jdbcType=LONGVARCHAR}
+      #{jobTask.versionName,jdbcType=VARCHAR},#{jobTask.runCommand,jdbcType=VARCHAR},#{jobTask.runSource,jdbcType=LONGVARCHAR},
+      #{jobTask.flowName,jdbcType=VARCHAR}
+
       )
     </foreach>
   </insert>
@@ -350,6 +359,9 @@
       <if test="runSource != null">
         run_source = #{runSource,jdbcType=LONGVARCHAR},
       </if>
+      <if test="runSource != null">
+        flow_name = #{flowName,jdbcType=VARCHAR},
+      </if>
     </set>
     where id = #{id,jdbcType=INTEGER}
   </update>

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

@@ -31,6 +31,7 @@
         <result column="version_name" jdbcType="VARCHAR" property="versionName"/>
         <result column="log_id" jdbcType="INTEGER" property="logId"/>
         <result column="run_command" jdbcType="VARCHAR" property="runCommand"/>
+        <result column="flow_name" jdbcType="VARCHAR" property="flowName"/>
     </resultMap>
 
     <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskSchedule">
@@ -42,7 +43,7 @@
     job_type, handler_name, node_desc, node_name, map_flow_id, node_timeout, is_virtual, 
     plugin_urls, priority, failed_retry_interval, routing_strategy, run_id, run_param, 
     run_source_desc, script_urls, source_principal, trigger_time, trigger_status, version_name, 
-    log_id, run_command
+    log_id, run_command,flow_name
   </sql>
     <sql id="Blob_Column_List">
     run_source
@@ -162,6 +163,9 @@
             <if test="runSource != null">
                 run_source,
             </if>
+            <if test="flowName != null">
+                flow_name,
+            </if>
         </trim>
         <trim prefix="values (" suffix=")" suffixOverrides=",">
             <if test="id != null">
@@ -251,6 +255,9 @@
             <if test="runSource != null">
                 #{runSource,jdbcType=LONGVARCHAR},
             </if>
+            <if test="flowName != null">
+                #{flowName,jdbcType=VARCHAR},
+            </if>
         </trim>
     </insert>
 
@@ -264,7 +271,7 @@
         run_id, run_param, run_source_desc,
         script_urls, source_principal, trigger_time,
         trigger_status, version_name, log_id,
-        run_command, run_source
+        run_command, run_source,flow_name
         )
         values
         <foreach collection="jobTaskSchedules" item="jobTaskSchedule" separator=",">
@@ -287,7 +294,8 @@
             #{jobTaskSchedule.triggerTime,jdbcType=BIGINT},
             #{jobTaskSchedule.triggerStatus,jdbcType=CHAR}, #{jobTaskSchedule.versionName,jdbcType=VARCHAR},
             #{jobTaskSchedule.logId,jdbcType=INTEGER},
-            #{jobTaskSchedule.runCommand,jdbcType=VARCHAR}, #{jobTaskSchedule.runSource,jdbcType=LONGVARCHAR}
+            #{jobTaskSchedule.runCommand,jdbcType=VARCHAR}, #{jobTaskSchedule.runSource,jdbcType=LONGVARCHAR},
+            #{jobTaskSchedule.flowName,jdbcType=VARCHAR}
             )
         </foreach>
     </insert>
@@ -380,6 +388,9 @@
             <if test="runSource != null">
                 run_source = #{runSource,jdbcType=LONGVARCHAR},
             </if>
+            <if test="flowName != null">
+                flow_name = #{flowName,jdbcType=VARCHAR},
+            </if>
         </set>
         where id = #{id,jdbcType=INTEGER}
     </update>

byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobThread.java

@@ -3,6 +3,7 @@ package com.byit.rpc;
 import com.alibaba.fastjson.JSON;
 import com.byit.job.dto.AdminSenPluginDto;
 import com.byit.job.dto.JobRunResultDto;
+import com.byit.job.enums.JobResultEnum;
 import com.byit.job.handler.interfaces.IJobHandler;
 import com.byit.job.vo.ReturnResult;
 import com.byit.utils.JobUtils;
@@ -41,18 +42,21 @@ public class RunJobThread implements Runnable {
         jobRunResultDto.setEndTime(new Date());
         jobRunResultDto.setLogId(adminSenPluginDto.getLogId());
         cn.hutool.http.HttpUtil.post(callbackUrl, JSON.toJSONString(jobRunResultDto));
-        log.info("---------服务器端:{}-------------", jobRunResultDto);
+        log.info("---------服务器端:{},花费时间：{}-------------", jobRunResultDto,jobRunResultDto.getStartTime().getTime()-jobRunResultDto.getEndTime().getTime());
     }
 
     private ReturnResult<String> runJob(String jobHandlerName,String param){
+        ReturnResult<String> returnResult=  ReturnResult.SUCCESS;
         Class<? extends IJobHandler> jobClass = JobUtils.jobCache.get(jobHandlerName);
         try {
             IJobHandler iJobHandler = jobClass.newInstance( );
-            return iJobHandler.execute(param);
+            returnResult =  iJobHandler.execute(param);
         } catch (Exception e) {
             e.printStackTrace( );
+            returnResult.setMsg(e.getMessage());
+            returnResult.setCode(JobResultEnum.FAIL.getRes());
         }
-        return null;
+        return returnResult;
 
     }
 }

byit-myth-job.properties

+path.variable.kotlin_bundled=D\:\\idea\\IntelliJ IDEA 2019.3.1\\plugins\\Kotlin\\kotlinc
+path.variable.maven_repository=E\:\\data\\repostory
+jdk.home.1.8=C\:/Program Files/Java/jdk1.8.0_121
+javac2.instrumentation.includeJavaRuntime=false
\ No newline at end of file

demo-client/byit-demo-client/pom.xml

@@ -43,6 +43,11 @@
             <artifactId>fastjson</artifactId>
             <version>1.2.62</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-exec</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -4,6 +4,8 @@ import com.byit.annotations.JobHandler;
 import com.byit.job.handler.BaseJobHandler;
 import com.byit.job.vo.ReturnResult;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * @program: byit-myth-job->DemoJob
  * @description: TODO
@@ -15,6 +17,10 @@ public class DemoJob extends BaseJobHandler {
     @Override
     public ReturnResult<String> execute(String s) throws Exception {
         System.out.println("--------------任务就这样运行了 addJob----------------"+s );
+        TimeUnit.SECONDS.sleep(10);
+        if("1".equals(s)){
+            throw new RuntimeException("我出异常了，信不信由你，我是爸爸");
+        }
         return ReturnResult.SUCCESS;
     }
 }

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

@@ -22,10 +22,10 @@ public class TestAddFlow {
 
     public static PluginFlow createFlow(){
         PluginFlow pluginFlow = new PluginFlow();
-        PluginFlowConfig build = PluginFlowConfig.builder().alarmEmail("huangfukexin@byitgroup.com,guominglei@byitgroup.com")
+        PluginFlowConfig build = PluginFlowConfig.builder().alarmEmail("huangfukexing@byitgroup.com")
                 .alarmlAction("1")
                 .execType("1")
-                .flowCron("0 0/5 * * * ? *")
+                .flowCron("0 0/1 * * * ? *")
                 .flowTimeout(TimeUnit.MINUTES.toMillis(30))
                 .priority("2")
                 .repeatCount(1)
@@ -33,7 +33,7 @@ public class TestAddFlow {
                 .build();
 
 
-        pluginFlow.setName("测试任务流");
+        pluginFlow.setName("测试时间执行的工作流");
         pluginFlow.setDesc("这是一个测试的任务流");
         pluginFlow.setConfig(build);
         pluginFlow.setPrincipal("皇甫科星");
@@ -73,7 +73,7 @@ public class TestAddFlow {
         pluginNode2.setAuthor("皇甫");
         pluginNode2.setJobType("JAVA");
         pluginNode2.setHandlerName("addJob");
-        pluginNode2.setRunParam("addJob1");
+        pluginNode2.setRunParam("1");
         pluginNodeConfig2.setFailedRetryCount(2);
         pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
         pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");

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

+package com.byit.job;
+
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.PumpStreamHandler;
+
+import java.io.*;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+public class TestPy {
+
+    public static void main(String[] args) throws IOException {
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        PumpStreamHandler pumpStreamHandler = new PumpStreamHandler(outputStream,outputStream,null);
+        DefaultExecutor defaultExecutor = new DefaultExecutor();
+        defaultExecutor.setExitValues(null);
+        defaultExecutor.setStreamHandler(pumpStreamHandler);
+        CommandLine commandline = new CommandLine("python");
+        commandline.addArgument("D:\\2020project\\byit-myth-job\\demo-client\\byit-demo-client\\src\\main\\resources\\test.py");
+        int execute = defaultExecutor.execute(commandline);
+
+        byte[] bytes = outputStream.toByteArray();
+        System.out.println(execute);
+        System.out.println(new String(bytes,0,bytes.length, StandardCharsets.UTF_8));
+    }
+}

demo-client/byit-demo-client/src/main/resources/test.py

+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+print("--------------------------------------------")
+print(100/10)
+
+print(5/0)
\ No newline at end of file

pom.xml

@@ -65,13 +65,19 @@
     <mysql-connector-java.version>5.1.47</mysql-connector-java.version>
     <springfox-swagger2.version>2.9.2</springfox-swagger2.version>
     <springfox-swagger-ui>2.9.2</springfox-swagger-ui>
-
+    <org-apache-commons>1.3</org-apache-commons>
 
 
     </properties>
 
     <dependencyManagement>
         <dependencies>
+            <!-- 外部脚本执行器 -->
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-exec</artifactId>
+                <version>${org-apache-commons}</version>
+            </dependency>
             <dependency>
                 <groupId>org.springframework.cloud</groupId>
                 <artifactId>spring-cloud-dependencies</artifactId>