修改部分功能

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/EmailEnum.java

+package com.byit.enums;
+
+/**
+ * @author Administrator
+ */
+
+public enum EmailEnum {
+    IS_ALARM_YES("0","已经告警"),
+    IS_ALARM_NO("1","没有告警")
+    ;
+    private String code;
+    private String msg;
+
+    EmailEnum(String code, String msg) {
+        this.code = code;
+        this.msg = msg;
+    }
+
+    public String getCode() {
+        return code;
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/NodeRunStatusPropertyEnum.java

+package com.byit.enums;
+
+/**
+ * 节点运行状态枚举
+ * @author huangfu
+ */
+public enum  NodeRunStatusPropertyEnum {
+    RUN_ING("0","运行中"),
+    RUN_SUCCESS("1","成功"),
+    RUN_FAILURE("2","失败"),
+    RE_RUN_SUCCESS("3","补批成功"),
+    RE_RUN_FAILURE("4","补批失败"),
+    KILL("5","kill")
+    ;
+    private String code;
+    private String msg;
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+
+    NodeRunStatusPropertyEnum(String code, String msg) {
+        this.code = code;
+        this.msg = msg;
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/RunRecordingEnum.java

@@ -15,6 +15,8 @@ public enum RunRecordingEnum {
     ,FLOW_STATUS_RUN_ING("2","工作流运行中")
     ,FLOW_STATUS_IS_STOP("3","工作流被暂停")
     ,FLOW_STATUS_IS_END("4","工作流已经完结")
+    ,FAIL_FAST_NO("0","不快速失败")
+    ,FAIL_FAST_YES("1","快速失败")
     ;
     private String code;
     private String message;

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

@@ -6,6 +6,13 @@ import org.apache.ibatis.annotations.Param;
 import java.util.List;
 
 public interface FlowMapper {
+
+    /**
+     * 根据ID查询
+     * @param id
+     * @return
+     */
+    Flow findFlowById(Integer id);
     /**
      * 查询半个小时内即将要执行的工作流
      * @param triggerNextTime

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

@@ -12,10 +12,9 @@ public interface NodeMapper {
     /**
      * 跟怒工作流ID和版本名称查询所有的节点
      * @param flowId
-     * @param versionName
      * @return
      */
-    List<Node> findNodeByFlowIdAndVersionName(@Param("flowId") Integer flowId,@Param("versionName") String versionName);
+    List<Node> findNodeByFlowIdAndVersionName(@Param("flowId") Integer flowId);
 
     int deleteById(Integer nodeId);
 

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

@@ -163,6 +163,10 @@ public class JobTaskRunLog implements Serializable {
      */
     @ApiModelProperty("重跑的运行标识")
     private String reRunId;
+    /**
+     * 日志文件的地址
+     */
+    private String logFileName;
 
     /**
      *

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

@@ -14,6 +14,12 @@ import java.util.List;
  */
 public interface FlowService {
     /**
+     * 根据ID查询
+     * @param id
+     * @return
+     */
+    Flow findFlowById(Integer id);
+    /**
      * 查询半个小时内即将要执行的工作流
      * @param triggerNextTime
      * @return

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

@@ -14,10 +14,9 @@ public interface NodeService {
     /**
      * 跟怒工作流ID和版本名称查询所有的节点
      * @param flowId
-     * @param versionName
      * @return
      */
-    List<Node> findNodeByFlowIdAndVersionName(Integer flowId,String versionName);
+    List<Node> findNodeByFlowIdAndVersionName(Integer flowId);
 
 
 

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

@@ -44,6 +44,11 @@ public class FlowServiceImpl implements FlowService {
     private NodeDependencyMapper nodeDependencyMapper;
 
     @Override
+    public Flow findFlowById(Integer id) {
+        return flowMapper.getById(id);
+    }
+
+    @Override
     public List<Flow> findHalfAnHourFlow(Long triggerNextTime) {
         return flowMapper.findHalfAnHourFlow(triggerNextTime);
     }

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

@@ -20,7 +20,7 @@ public class NodeServiceImpl implements NodeService {
     private NodeMapper nodeMapper;
 
     @Override
-    public List<Node> findNodeByFlowIdAndVersionName(Integer flowId, String versionName) {
-        return nodeMapper.findNodeByFlowIdAndVersionName(flowId,versionName);
+    public List<Node> findNodeByFlowIdAndVersionName(Integer flowId) {
+        return nodeMapper.findNodeByFlowIdAndVersionName(flowId);
     }
 }

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

+package com.byit.service.mapservice;
+
+import com.byit.model.JobTask;
+
+import java.net.UnknownHostException;
+
+/**
+ * @author Administrator
+ */
+public interface RunRecordingAndJobTaskService {
+    /**
+     * 保存节点日志
+     * 保存运行记录
+     * 保存任务节点
+     * @param jobTask
+     * @throws Exception
+     */
+    void saveRunRecordingAndTask(JobTask jobTask) throws Exception;
+}

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

-package com.byit.service;
+package com.byit.service.mapservice;
 
 import com.byit.model.JobTask;
 import org.springframework.transaction.annotation.Transactional;

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

+package com.byit.service.mapservice.impl;
+
+import com.byit.enums.EmailEnum;
+import com.byit.enums.FlowPropertyEnum;
+import com.byit.enums.NodeRunStatusPropertyEnum;
+import com.byit.enums.RunRecordingEnum;
+import com.byit.model.*;
+import com.byit.service.*;
+import com.byit.service.mapservice.RunRecordingAndJobTaskService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 运行记录与任务表的事务处理
+ * @author huangfu
+ */
+@Service
+@Slf4j
+@Transactional
+public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTaskService {
+    private final JobTaskRunLogService jobTaskRunLogService;
+    private final NodeService nodeService;
+    private final FlowService flowService;
+    private final RunRecordingService runRecordingService;
+    private final JobTaskService jobTaskService;
+
+    public RunRecordingAndJobTaskServiceImpl(JobTaskRunLogService jobTaskRunLogService, NodeService nodeService, FlowService flowService, RunRecordingService runRecordingService, JobTaskService jobTaskService) {
+        this.jobTaskRunLogService = jobTaskRunLogService;
+        this.nodeService = nodeService;
+        this.flowService = flowService;
+        this.runRecordingService = runRecordingService;
+        this.jobTaskService = jobTaskService;
+    }
+    /**
+     * 保存节点日志
+     * 保存运行记录
+     * 保存任务节点
+     * @param jobTask
+     * @throws Exception
+     */
+    @Override
+    public void saveRunRecordingAndTask(JobTask jobTask) throws UnknownHostException {
+        log.info("-----------saveRunRecordingAndTask start【虚节点保存服务】--------------");
+        //根据 map_flow_id查询当前的版本的工作流  使用祝工作流的runId 保存到执行记录表和任务表
+        Integer mapFlowId = jobTask.getMapFlowId();
+        //主工作流
+        Flow mainFlow = flowService.findFlowById(jobTask.getFlowId());
+        //内嵌工作流
+        Flow virFlow = flowService.findFlowById(mapFlowId);
+        //查看是否跟随工作流
+        boolean equals = "1".equals(mainFlow.getScheduleFollow());
+        boolean virFlag = "1".equals(virFlow.getScheduleFollow());
+        //在日志表里面创建一条记录
+        log.info("-----------【虚节点保存日志服务】--------------");
+        JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+        jobTaskRunLog.setRunId(jobTask.getRunId());
+        jobTaskRunLog.setNodeId(jobTask.getNodeId());
+        jobTaskRunLog.setNodeName(jobTask.getNodeName());
+        jobTaskRunLog.setJobType(jobTask.getJobType());
+        jobTaskRunLog.setFlowId(mapFlowId);
+        jobTaskRunLog.setFailedRemainingCount(jobTask.getFailedRetryCount());
+        jobTaskRunLog.setVersionName(jobTask.getVersionName());
+        jobTaskRunLog.setFlowName(jobTask.getFlowName());
+        jobTaskRunLog.setIsVirtual(jobTask.getIsVirtual());
+        jobTaskRunLog.setMapFlowId(jobTask.getMapFlowId());
+        jobTaskRunLog.setRunCode(NodeRunStatusPropertyEnum.RUN_ING.getCode());
+        jobTaskRunLog.setRunType(jobTask.getJobType());
+        Date thisDate = new Date();
+        jobTaskRunLog.setTriggerTime(thisDate);
+        jobTaskRunLog.setStartTime(thisDate);
+        jobTaskRunLog.setAlertEnd("0");
+        jobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
+        log.info("-----------【虚节点保存运行记录】--------------");
+        //保存进运行记录表
+        RunRecording runRecording = RunRecording.builder()
+                .runId(jobTask.getRunId())
+                .flowId(mapFlowId)
+                .flowName(virFlow.getFlowName())
+                .flowVersionName(virFlow.getVersionName())
+                .flowStatus("2")
+                .flowTimeout(virFlow.getFlowTimeout())
+                .dispatchIp(InetAddress.getLocalHost().getHostAddress())
+                .alarmEmail(virFlow.getAlarmEmail())
+                .alarmlAction(virFlow.getAlarmlAction())
+                .priority(virFlow.getPriority())
+                .triggerTime(equals?mainFlow.getTriggerNextTime():virFlow.getTriggerNextTime())
+                .principal(virFlow.getPrincipal())
+                .startTime(new Date())
+                .isAlarm(EmailEnum.IS_ALARM_NO.getCode())
+                .isInner(FlowPropertyEnum.IS_INNER.getCode())
+                .failFast(RunRecordingEnum.FAIL_FAST_NO.getCode()).build();
+        runRecordingService.saveRunRecording(runRecording);
+        log.info("-----------【虚节点对应节点保存到任务表】--------------");
+        //获取所有的节点，开始将所有节点保存到任务表
+        List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(mapFlowId);
+        List<JobTask> jobTasks = nodeByFlowIdAndVersionName.stream()
+                .map(node -> {
+                    JobTask task = new JobTask();
+                    BeanUtils.copyProperties(node, task);
+                    if (equals) {
+                        task.setTriggerTime(mainFlow.getTriggerNextTime());
+                    } else if (virFlag) {
+                        task.setTriggerTime(virFlow.getTriggerNextTime());
+                    }
+                    task.setRunId(jobTask.getRunId());
+                    return task;
+                }).collect(Collectors.toList());
+        jobTaskService.saveJobTasks(jobTasks);
+        log.info("-----------saveRunRecordingAndTask end【虚节点保存服务】--------------");
+    }
+}

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

-package com.byit.service.impl;
+package com.byit.service.mapservice.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 com.byit.service.mapservice.TaskAndLogServer;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;

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

@@ -70,7 +70,7 @@ public class FlowScanHelper {
                             log.debug("-----------------【工作流{}的执行次数大于0，放行】-------------------------",flow.getFlowName());
                             String versionName = flow.getVersionName();
                             Integer flowId = flow.getFlowId();
-                            List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId, versionName);
+                            List<Node> nodeByFlowIdAndVersionName = nodeService.findNodeByFlowIdAndVersionName(flowId);
                             if(CollectionUtil.isNotEmpty(nodeByFlowIdAndVersionName)){
                                 runNodeServer.saveRunRecAndTask(flow,nodeByFlowIdAndVersionName);
                                 flow.setRemainingCount(flow.getRemainingCount()-1);

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

@@ -2,12 +2,11 @@ package com.byit.thread;
 
 import cn.hutool.core.collection.CollectionUtil;
 import com.byit.job.WorkRoulette;
-import com.byit.model.JobTask;
-import com.byit.model.JobTaskRunLog;
-import com.byit.model.JobTaskRunLogWithBLOBs;
-import com.byit.model.JobTaskSchedule;
+import com.byit.model.*;
 import com.byit.service.*;
 import com.byit.service.impl.JobTaskRunLogServiceImpl;
+import com.byit.service.mapservice.RunRecordingAndJobTaskService;
+import com.byit.service.mapservice.TaskAndLogServer;
 import com.byit.task.JavaBeanJobTask;
 import com.byit.util.SpringUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -37,12 +36,14 @@ public class JobScheduleHelper{
     private DataSource dataSource;
 
 
+    private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
     private final JobTaskService jobTaskService;
     private final JobTaskScheduleService jobTaskScheduleService;
     private final NodeDependencyService nodeDependencyService;
     private final JobTaskRunLogService jobTaskRunLogService;
     private final TaskAndLogServer taskAndLogServer;
 
+
     /**
      * 读取任务节点的预读
      */
@@ -70,7 +71,8 @@ public class JobScheduleHelper{
     private volatile boolean scheduleThreadToStop = false;
 
     @Autowired
-    public JobScheduleHelper(JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer) {
+    public JobScheduleHelper(RunRecordingAndJobTaskService runRecordingAndJobTaskService, JobTaskScheduleService jobTaskScheduleService, JobTaskService jobTaskService, NodeDependencyService nodeDependencyService, JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer) {
+        this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
         this.jobTaskScheduleService = jobTaskScheduleService;
         this.jobTaskService = jobTaskService;
         this.nodeDependencyService = nodeDependencyService;
@@ -128,8 +130,13 @@ public class JobScheduleHelper{
                         jobTasks.forEach(jobTask -> {
                             //虚节点的状态
                             if("0".equals(jobTask.getIsVirtual())){
-                                //在日志表里面创建一条记录
-                                //根据 map_flow_id查询当前的版本的工作流  使用祝工作流的runId 保存到执行记录表和任务表
+                                try {
+                                    runRecordingAndJobTaskService.saveRunRecordingAndTask(jobTask);
+                                    log.info("----------------【虚节点保存成功，删除虚节点】--------------------");
+                                    jobTaskService.removeMythJobTaskById(jobTask.getId());
+                                } catch (Exception e) {
+                                    e.printStackTrace();
+                                }
                             }else{
                                 if("start".equals(jobTask.getNodeName()) ){
                                     log.debug("任务：{}", jobTask);

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

@@ -37,7 +37,12 @@
   <select id="findHalfAnHourFlow" resultMap="BaseResultMap">
     select
     <include refid="Base_Column_List" />
-    from flow where trigger_next_time <![CDATA[ <= ]]> #{triggerNextTime,jdbcType=BIGINT} and remaining_count != 0
+    from flow
+    where
+      trigger_next_time <![CDATA[ <= ]]> #{triggerNextTime,jdbcType=BIGINT}
+      and remaining_count != 0
+      and start_up = '0'
+      and is_inner = '1'
   </select>
 
   <select id="getById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
@@ -55,6 +60,12 @@
     where workspace_id = #{workspaceId}
         and flow_name = #{flowName}
   </select>
+    <select id="findFlowById" resultType="com.byit.model.Flow">
+      select
+      <include refid="Base_Column_List" />
+      from flow
+      where flow_id = #{id,jdbcType=INTEGER}
+    </select>
 
     <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2019-12-31 -->

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

@@ -26,6 +26,7 @@
     <result column="job_type" jdbcType="CHAR" property="jobType" />
     <result column="run_id" jdbcType="VARCHAR" property="runId" />
     <result column="re_run_id" jdbcType="VARCHAR" property="reRunId" />
+    <result column="log_file_name" jdbcType="VARCHAR" property="logFileName" />
   </resultMap>
   <resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
     <result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
@@ -35,6 +36,7 @@
     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,run_id,re_run_id
+    ,log_file_name
   </sql>
   <sql id="Blob_Column_List">
     run_msg, trigger_msg
@@ -167,6 +169,9 @@
       <if test="reRunId != null">
         re_run_id,
       </if>
+      <if test="logFileName != null">
+        log_file_name,
+      </if>
     </trim>
     <trim prefix="values (" suffix=")" suffixOverrides=",">
       <if test="logId != null">
@@ -247,6 +252,9 @@
       <if test="reRunId != null">
         #{reRunId,jdbcType=VARCHAR},
       </if>
+      <if test="logFileName != null">
+        #{logFileName,jdbcType=VARCHAR},
+      </if>
     </trim>
   </insert>
   <update id="updateJobTaskRunLogWithBLOBs" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
@@ -327,6 +335,9 @@
       <if test="reRunId != null">
         re_run_id = #{reRunId,jdbcType=VARCHAR},
       </if>
+      <if test="logFileName != null">
+        log_file_name = #{logFileName,jdbcType=VARCHAR},
+      </if>
     </set>
     where log_id = #{logId,jdbcType=INTEGER}
   </update>
@@ -402,6 +413,9 @@
       <if test="reRunId != null">
         re_run_id = #{reRunId,jdbcType=VARCHAR},
       </if>
+      <if test="logFileName != null">
+        log_file_name = #{logFileName,jdbcType=VARCHAR},
+      </if>
     </set>
     where log_id = #{logId,jdbcType=INTEGER}
   </update>

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

@@ -61,7 +61,7 @@
     ,
     <include refid="Blob_Column_List" />
     from node
-    where flow_id = #{flowId,jdbcType=INTEGER} AND version_name = #{versionName,jdbcType=VARCHAR}
+    where flow_id = #{flowId,jdbcType=INTEGER}
   </select>
 
   <select id="getById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">

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

@@ -203,7 +203,7 @@
         #{isInner,jdbcType=CHAR},
       </if>
       <if test="failFast != null">
-        #{fail_fast,jdbcType=CHAR},
+        #{failFast,jdbcType=CHAR},
       </if>
     </trim>
   </insert>

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

@@ -32,9 +32,8 @@ public class TestAddFlow {
                 .scheduleFollow("1")
                 .build();
 
-
-        pluginFlow.setName("自动化测试原子弹");
-        pluginFlow.setDesc("自动化测试原子弹");
+        pluginFlow.setName("虚拟节点");
+        pluginFlow.setDesc("虚拟节点");
         pluginFlow.setConfig(build);
         pluginFlow.setPrincipal("皇甫科星");
         pluginFlow.setRePublish(false);

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

+package com.byit.job;
+
+import com.byit.job.dto.plugin.*;
+import com.byit.rpc.remoting.invoker.route.LoadBalance;
+import com.byit.utils.JobUtils;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class TestAddFlow2 {
+
+
+    public static void main(String[] args) {
+        PluginPackage pluginPackage = new PluginPackage();
+        pluginPackage.setWorkspaceName("test");
+        pluginPackage.setFlow(createFlow());
+        JobUtils.publish(pluginPackage);
+
+    }
+
+
+    public static PluginFlow createFlow(){
+        PluginFlow pluginFlow = new PluginFlow();
+        PluginFlowConfig build = PluginFlowConfig.builder().alarmEmail("huangfukexing@byitgroup.com")
+                .alarmlAction("1")
+                .execType("1")
+                .flowCron("0 0/1 * * * ? *")
+                .flowTimeout(TimeUnit.MINUTES.toMillis(30))
+                .priority("2")
+                .repeatCount(1)
+                .scheduleFollow("1")
+                .build();
+
+
+        pluginFlow.setName("自动化测试原子弹");
+        pluginFlow.setDesc("自动化测试原子弹");
+        pluginFlow.setConfig(build);
+        pluginFlow.setPrincipal("皇甫科星");
+        pluginFlow.setRePublish(false);
+        pluginFlow.setAuthor("huangfusuper");
+        pluginFlow.setNodeList(createNodes());
+        return pluginFlow;
+    }
+
+    /**
+     * 创建节点
+     * @return
+     */
+    public static List<PluginBaseNode> createNodes(){
+        PluginNode pluginNode1 = new PluginNode();
+        PluginNodeConfig pluginNodeConfig1 = new PluginNodeConfig();
+        pluginNode1.setName("start");
+        pluginNode1.setDesc("我是开始节点，打死你");
+        pluginNode1.setType("node");
+        pluginNode1.setAuthor("郭郭");
+        pluginNode1.setJobType("JAVA");
+        pluginNode1.setHandlerName("start");
+        pluginNodeConfig1.setFailedRetryCount(2);
+        pluginNodeConfig1.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
+        pluginNodeConfig1.setNodeCron("0 0/7 * * * ? *");
+        pluginNodeConfig1.setNodeTimeout(-1L);
+        pluginNodeConfig1.setPluginUrls("http://127.0.0.1:8888");
+        pluginNodeConfig1.setPriority("2");
+        pluginNodeConfig1.setRoutingStrategy(LoadBalance.ROUND.name());
+        pluginNode1.setConfig(pluginNodeConfig1);
+
+        PluginNode pluginNode2 = new PluginNode();
+        PluginNodeConfig pluginNodeConfig2 = new PluginNodeConfig();
+        pluginNode2.setName("中间节点1");
+        pluginNode2.setDesc("我是中间节点1，我依赖于开始节点");
+        pluginNode2.setType("node");
+        pluginNode2.setAuthor("皇甫");
+        pluginNode2.setJobType("JAVA");
+        pluginNode2.setHandlerName("addJob");
+        pluginNode2.setRunParam("add1");
+        pluginNodeConfig2.setFailedRetryCount(2);
+        pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
+        pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
+        pluginNodeConfig2.setNodeTimeout(-1L);
+        pluginNodeConfig2.setPluginUrls("http://127.0.0.1:8888");
+        pluginNodeConfig2.setPriority("2");
+        pluginNodeConfig2.setRoutingStrategy(LoadBalance.ROUND.name());
+        pluginNode2.setConfig(pluginNodeConfig2);
+        pluginNode2.setDependNodeNameList(Collections.singletonList("start"));
+
+        PluginFlow flow = TestAddFlow.createFlow();
+        flow.setDependNodeNameList(Collections.singletonList("start"));
+        flow.setType("flow");
+
+        PluginNode pluginNode13 = new PluginNode();
+        PluginNodeConfig pluginNodeConfig13 = new PluginNodeConfig();
+        pluginNode13.setName("中间节点2");
+        pluginNode13.setDesc("我是中间节点2，我依赖中间节点1");
+        pluginNode13.setType("node");
+        pluginNode13.setAuthor("皇甫");
+        pluginNode13.setJobType("JAVA");
+        pluginNode13.setHandlerName("addJob");
+        pluginNode13.setRunParam("addJob2");
+        pluginNodeConfig13.setFailedRetryCount(2);
+        pluginNodeConfig13.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
+        pluginNodeConfig13.setNodeCron("0 0/7 * * * ? *");
+        pluginNodeConfig13.setNodeTimeout(-1L);
+        pluginNodeConfig13.setPluginUrls("http://127.0.0.1:8888");
+        pluginNodeConfig13.setPriority("2");
+        pluginNodeConfig13.setRoutingStrategy(LoadBalance.ROUND.name());
+        pluginNode13.setConfig(pluginNodeConfig13);
+        pluginNode13.setDependNodeNameList(Collections.singletonList("中间节点1"));
+
+        PluginNode pluginNode4 = new PluginNode();
+        PluginNodeConfig pluginNodeConfig4 = new PluginNodeConfig();
+        pluginNode4.setName("中间节点3");
+        pluginNode4.setDesc("我是中间节点3，我依赖于中间节点2");
+        pluginNode4.setType("node");
+        pluginNode4.setAuthor("皇甫");
+        pluginNode4.setJobType("JAVA");
+        pluginNode4.setHandlerName("addJob");
+        pluginNode4.setRunParam("add3success");
+        pluginNodeConfig4.setFailedRetryCount(2);
+        pluginNodeConfig4.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
+        pluginNodeConfig4.setNodeCron("0 0/7 * * * ? *");
+        pluginNodeConfig4.setNodeTimeout(-1L);
+        pluginNodeConfig4.setPluginUrls("http://127.0.0.1:8888");
+        pluginNodeConfig4.setPriority("2");
+        pluginNodeConfig4.setRoutingStrategy(LoadBalance.ROUND.name());
+        pluginNode4.setConfig(pluginNodeConfig4);
+        pluginNode4.setDependNodeNameList(Collections.singletonList("中间节点2"));
+
+        PluginNode pluginNode5 = new PluginNode();
+        PluginNodeConfig pluginNodeConfig5 = new PluginNodeConfig();
+        pluginNode5.setName("end");
+        pluginNode5.setDesc("我是结束节点，我依赖于所有的末尾节点");
+        pluginNode5.setType("node");
+        pluginNode5.setAuthor("皇甫");
+        pluginNode5.setJobType("JAVA");
+        pluginNode5.setHandlerName("END");
+        pluginNode5.setRunParam("END");
+        pluginNodeConfig5.setFailedRetryCount(2);
+        pluginNodeConfig5.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
+        pluginNodeConfig5.setNodeCron("0 0/7 * * * ? *");
+        pluginNodeConfig5.setNodeTimeout(-1L);
+        pluginNodeConfig5.setPluginUrls("http://127.0.0.1:8888");
+        pluginNodeConfig5.setPriority("2");
+        pluginNodeConfig5.setRoutingStrategy(LoadBalance.ROUND.name());
+        pluginNode5.setConfig(pluginNodeConfig5);
+        pluginNode5.setDependNodeNameList(Arrays.asList("中间节点3","虚拟节点"));
+
+
+        return Arrays.asList(pluginNode5, pluginNode4, pluginNode13, pluginNode2, pluginNode1,flow);
+    }
+
+
+
+}

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

@@ -5,46 +5,34 @@ import org.apache.commons.exec.DefaultExecutor;
 import org.apache.commons.exec.ExecuteWatchdog;
 import org.apache.commons.exec.PumpStreamHandler;
 
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
+import java.io.*;
+import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
 
 public class TestPy {
 
     public static void main(String[] args) throws IOException, InterruptedException {
-
-        List<ExecuteWatchdog> watchdogList = new ArrayList<>();
-
-        Thread thread = new Thread(() -> {
         ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
         PumpStreamHandler pumpStreamHandler = new PumpStreamHandler(outputStream,outputStream,null);
         ExecuteWatchdog watchdog = new ExecuteWatchdog(Integer.MAX_VALUE);
-            watchdogList.add(watchdog);
         DefaultExecutor defaultExecutor = new DefaultExecutor();
         defaultExecutor.setWatchdog(watchdog);
         defaultExecutor.setExitValues(null);
         defaultExecutor.setStreamHandler(pumpStreamHandler);
         CommandLine commandline = new CommandLine("python");
-            commandline.addArgument("D:\\workspace\\pycharmWorkSpace\\test\\com.test\\Test.py");
-            int execute = 0;
+        commandline.addArgument("D:\\2020project\\byit-myth-job\\demo-client\\byit-demo-client\\test.py");
+        new Thread(() ->{
             try {
-                execute = defaultExecutor.execute(commandline);
+                int execute = defaultExecutor.execute(commandline);
             } catch (IOException e) {
                 e.printStackTrace();
             }
-            byte[] bytes = outputStream.toByteArray();
-            System.out.println(execute);
-
-            System.out.println(new String(bytes,0,bytes.length, StandardCharsets.UTF_8));
         });
-        thread.start();
-
-        Thread.sleep(10000);
-        watchdogList.forEach(watchdog ->{
+        byte[] bytes = outputStream.toByteArray();
+        Thread.sleep(3000);
         watchdog.destroyProcess();
-        });
 
+
+        System.out.println(new String(bytes,0,bytes.length, StandardCharsets.UTF_8));
     }
 }

demo-client/byit-demo-client/test.py

 import time
 
 while True :
-    print("--------------------------------------------")
+    time.sleep(1)
+    print("头给你打烂")