执行机添加杀死节点执行线程的接口

工作流管理工具添加杀死节点的方法

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

 package com.byit.util;
 
+import cn.hutool.http.HttpUtil;
+import com.alibaba.fastjson.JSON;
+import com.byit.job.vo.ResponseResult;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.HashMap;
+import java.util.Map;
 
 /**
  * @Description 工作流上下文内容
@@ -14,7 +18,7 @@ public class FlowContent {
     //工作流名称+运行的runid
     private String flowkey;
     //任务的运行日志id和具体的运行日志的机器的访问url
-    private HashMap<String, String> jobExectUrl = new HashMap<>(30);
+    private HashMap<Integer, String> jobExectUrl = new HashMap<>(30);
     //总节点数目
     private Integer allJobNum;
     //已经触发调度的节点数目
@@ -35,21 +39,41 @@ public class FlowContent {
     }
 
     public synchronized Boolean killFlow(){
-
-        return false;
+        Boolean result = true;
+        for (Integer logId : jobExectUrl.keySet()){
+            String exectUrl = jobExectUrl.get(logId);
+            if (!killJob(logId, exectUrl)){
+                result = false;
+            }
+        }
+        return result;
     }
 
-    public synchronized Boolean killJob(String logId){
+    public synchronized Boolean killJob(Integer logId){
         String exectUrl = jobExectUrl.get(logId);
         if (null != exectUrl && !"".equals(exectUrl)){
             log.info("{} -- {} 调用杀死命令", flowkey, logId);
-            //具体访问的URL
+            return killJob(logId, exectUrl);
         }
         log.error("{}下未查到{}的运行日志", flowkey, logId);
         return false;
     }
 
-    public synchronized void putJob(String logId, String exectUrl){
+    private synchronized Boolean killJob(Integer logId, String exectUrl){
+        //具体访问的URL
+        String killUrl = exectUrl + "/processManager/killJob";
+        Map<String, Integer> requestMap = new HashMap<>(2);
+        requestMap.put("logId", logId);
+        String killResult = HttpUtil.post(killUrl, JSON.toJSONString(requestMap));
+        log.info("请求结果{}", killResult);
+        ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
+        if ("SUCCESS".equals(responseResult.getResult())){
+            return true;
+        }
+        return false;
+    }
+
+    public synchronized void putJob(Integer logId, String exectUrl){
         jobExectUrl.put(logId, exectUrl);
         //已经触发的节点数目加1
         if (targetJobNum == null || targetJobNum < 0) {
@@ -78,4 +102,27 @@ public class FlowContent {
 
         log.error("{}下未查到{}的运行日志", flowkey, logId);
     }
+
+    //获取工作流下所有节点数目
+    public Integer getAllJobNum(){
+        return allJobNum;
+    }
+
+    //获取已经触发的工作流下节点数目
+    public Integer getTargetJobNum(){
+        return targetJobNum;
+    }
+
+    /**
+     * 获取对应的日志id的执行地址
+     * @param logId
+     * @return
+     */
+    public synchronized String getExectUrl(Integer logId){
+        String exectUrl = jobExectUrl.get(logId);
+        if (null != null && !"".equals(exectUrl)){
+            return exectUrl;
+        }
+        return  null;
+    }
 }

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

 package com.byit.executor.jobExecutor.process;
 
-import com.byit.executor.conf.FileSystemContext;
 import com.byit.executor.util.JobContentUtil;
 import com.byit.executor.util.LogGobbler;
 import com.google.common.base.Joiner;

byit-myth-executor/myth-executor-server/pom.xml

@@ -28,17 +28,12 @@
             <groupId>myth-job</groupId>
             <artifactId>byit-myth-rpc</artifactId>
         </dependency>
-
-       <!-- &lt;!&ndash; slf4j &ndash;&gt;
         <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
+            <groupId>com.byit</groupId>
+            <artifactId>byit-validation-starter</artifactId>
+            <version>1.0-SNAPSHOT</version>
         </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-            <scope>test</scope>
-        </dependency>-->
+
     </dependencies>
 
 </project>
\ No newline at end of file

byit-myth-executor/myth-executor-server/src/main/java/com/byit/ExecutorServiceApplication.java

@@ -15,3 +15,4 @@ public class ExecutorServiceApplication {
         SpringApplication.run(ExecutorServiceApplication.class,args);
     }
 }
+

byit-myth-executor/myth-executor-server/src/main/java/com/byit/controller/ProcessManagerController.java

+package com.byit.controller;
+
+import com.byit.executor.jobExecutor.process.MythJobProcess;
+import com.byit.executor.util.JobContentUtil;
+import com.byit.utils.ValidationUtil;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @Description
+ * @Author guo_m
+ * @Date 2020-02-22
+ */
+@RestController("processManager")
+public class ProcessManagerController {
+
+    @PostMapping("killJob")
+    @ResponseBody
+    public String killJob(Integer logId){
+        ValidationUtil.dataNotNull(logId, "日志id不允许为空！");
+        MythJobProcess mythJobProcess = JobContentUtil.getJobThread(logId);
+        ValidationUtil.dataNotNull(mythJobProcess, logId + "已经运行结束或还未调用！");
+        mythJobProcess.hardKill();
+        return "SUCESS";
+    }
+}