工作流上下文管理

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

+package com.byit.util;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+
+/**
+ * @Description 工作流上下文内容
+ * @Author guo_m
+ * @Date 2020-02-19
+ */
+@Slf4j
+public class FlowContent {
+    //工作流名称+运行的runid
+    private String flowkey;
+    //任务的运行日志id和具体的运行日志的机器的访问url
+    private HashMap<String, String> jobExectUrl = new HashMap<>(30);
+    //总节点数目
+    private Integer allJobNum;
+    //已经触发调度的节点数目
+    private Integer targetJobNum;
+    //未调度的节点数目
+    private Integer unRunIngJobNum;
+    //运行完成的节点数目
+    private Integer finshJobNUm;
+
+    public FlowContent(){
+
+    }
+
+    public FlowContent(String flowkey, Integer allJobNum){
+        this.flowkey = flowkey;
+        this.allJobNum = allJobNum;
+        this.unRunIngJobNum = allJobNum;
+    }
+
+    public synchronized Boolean killFlow(){
+
+        return false;
+    }
+
+    public synchronized Boolean killJob(String logId){
+        String exectUrl = jobExectUrl.get(logId);
+        if (null != exectUrl && !"".equals(exectUrl)){
+            log.info("{} -- {} 调用杀死命令", flowkey, logId);
+            //具体访问的URL
+        }
+        log.error("{}下未查到{}的运行日志", flowkey, logId);
+        return false;
+    }
+
+    public synchronized void putJob(String logId, String exectUrl){
+        jobExectUrl.put(logId, exectUrl);
+        //已经触发的节点数目加1
+        if (targetJobNum == null || targetJobNum < 0) {
+            targetJobNum = 1;
+        } else {
+            targetJobNum++;
+        }
+        //未触发的节点数目减1
+        if (unRunIngJobNum != null && unRunIngJobNum > 0) {
+            targetJobNum--;
+        }
+    }
+
+
+    public synchronized void finishJob(String logId){
+        String exectUrl = jobExectUrl.remove(logId);
+        if (null != exectUrl && !"".equals(exectUrl)){
+            //完成的节点数目加1
+            if (finshJobNUm == null || finshJobNUm < 0) {
+                finshJobNUm = 1;
+            } else {
+                finshJobNUm++;
+            }
+            log.info("{} -- {} 执行完毕", flowkey, logId);
+        }
+
+        log.error("{}下未查到{}的运行日志", flowkey, logId);
+    }
+}

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

+package com.byit.util;
+
+import java.util.HashMap;
+
+/**
+ * @Description 工作流上下文内容工具
+ * @Author guo_m
+ * @Date 2020-02-19
+ */
+public class FlowContentManager {
+    //存储当前的调度中心运行了哪些工作流(key为工作流名称加runid)
+    private static HashMap<String, FlowContent> flowContentHashMap = new HashMap<>(20);
+
+    public static synchronized void putFlowContent(String flowKey, FlowContent flowContent){
+        flowContentHashMap.put(flowKey, flowContent);
+    }
+
+    public static synchronized FlowContent remove(String flowKey){
+        FlowContent flowContent = flowContentHashMap.remove(flowKey);
+        return flowContent;
+    }
+
+    public static synchronized FlowContent getFlowContent(String flowKey){
+        FlowContent flowContent = flowContentHashMap.get(flowKey);
+        return flowContent;
+    }
+
+    public static synchronized FlowContent createFlowContent(String flowKey, Integer allJobNum){
+        FlowContent flowContent = new FlowContent(flowKey, allJobNum);
+        return flowContent;
+    }
+
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/Test.java

@@ -20,23 +20,24 @@ public class Test {
 
         Thread thread = new Thread(() -> {
             List<String> cmdList = new ArrayList<>();
-            cmdList = Arrays.asList("python D:\\workspace\\pycharmWorkSpace\\test\\com.test\\Test.py".split(" "));
+            cmdList = Arrays.asList("python D:\\workSpace\\python\\test\\test1.py".split(" "));
             Map<String, String> env = new HashMap<>();
-            env.put("python", "C:\\Program Files\\Python38");
-            MythJobProcess mythJobProcess = new MythJobProcess(cmdList, env, "D:\\workspace\\pycharmWorkSpace\\test\\com.test");
+//            env.put("python", "C:\\Program Files\\Python38");
+            MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, 1);
             list.add(mythJobProcess);
+            new Thread(mythJobProcess).start();
             try {
-                String log = mythJobProcess.run();
-                System.out.println("运行日志" + log);
-            } catch (IOException e) {
+                Thread.sleep(3000);
+            } catch (InterruptedException e) {
                 e.printStackTrace();
             }
+            System.out.println("运行日志" + mythJobProcess.getLogContent());
         });
 
         thread.start();
 
 
-        Thread.sleep(30000);
+        Thread.sleep(4000);
 
         list.forEach(process -> {
             process.hardKill();

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/MythJobProcess.java

@@ -3,6 +3,7 @@ package com.byit.executor.jobExecutor.process;
 import com.byit.executor.util.JobContentUtil;
 import com.byit.executor.util.LogGobbler;
 import com.google.common.base.Joiner;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.File;
@@ -22,7 +23,7 @@ import java.util.concurrent.TimeUnit;
  * @create: 2020-01-13 17:56
  */
 @Slf4j
-public class MythJobProcess {
+public class MythJobProcess implements Runnable{
     //杀死命令
     public static String KILL_COMMAND = "kill";
     //工作目录
@@ -47,6 +48,8 @@ public class MythJobProcess {
     private String effectiveUser = null;
     //本次任务对应的日志id
     private Integer logId;
+    //执行日志内容
+    private String logContent;
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
                           final String workingDir, final Integer logId) {
@@ -70,7 +73,9 @@ public class MythJobProcess {
     /**
      * 执行此过程，直到完成为止
      */
-    public String run() throws IOException {
+    @SneakyThrows
+    @Override
+    public void run() {
         //判断是否执行过
         if (this.isStarted() || this.isComplete()) {
             throw new IllegalStateException("该过程只能使用一次");
@@ -136,7 +141,8 @@ public class MythJobProcess {
             if (this.process.isAlive()){
                 this.process.destroy();
             }
-            return outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
+            logContent = outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
+            //TODO 将日志写到fastdfs上
         }
     }
 
@@ -291,4 +297,8 @@ public class MythJobProcess {
     public String getEffectiveUser() {
         return this.effectiveUser;
     }
+
+    public String getLogContent(){
+        return logContent;
+    }
 }