Merge remote-tracking branch 'origin/developer' into developer

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-admin-core/src/main/java/com/byit/util/FlowContentManager.java

@@ -11,20 +11,41 @@ public class FlowContentManager {
     //存储当前的调度中心运行了哪些工作流(key为工作流名称加runid)
     private static HashMap<String, FlowContent> flowContentHashMap = new HashMap<>(20);
 
+    /**
+     * 放置工作流运行内容
+     * @param flowKey
+     * @param flowContent
+     */
     public static synchronized void putFlowContent(String flowKey, FlowContent flowContent){
         flowContentHashMap.put(flowKey, flowContent);
     }
 
+    /**
+     * 移除工作流运行内容
+     * @param flowKey
+     * @return
+     */
     public static synchronized FlowContent remove(String flowKey){
         FlowContent flowContent = flowContentHashMap.remove(flowKey);
         return flowContent;
     }
 
+    /**
+     * 获取工作流运行内容
+     * @param flowKey
+     * @return
+     */
     public static synchronized FlowContent getFlowContent(String flowKey){
         FlowContent flowContent = flowContentHashMap.get(flowKey);
         return flowContent;
     }
 
+    /**
+     * 创建工作流运行内容主体
+     * @param flowKey
+     * @param allJobNum
+     * @return
+     */
     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/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;
@@ -15,6 +14,7 @@ import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit;
  * @create: 2020-01-13 17:56
  */
 @Slf4j
-public class MythJobProcess implements Runnable{
+public class MythJobProcess implements Callable<String>{
     //杀死命令
     public static String KILL_COMMAND = "kill";
     //工作目录
@@ -49,10 +49,6 @@ public class MythJobProcess implements Runnable{
     private String effectiveUser = null;
     //本次任务对应的日志id
     private Integer logId;
-    //执行日志内容
-    private String logFilePath;
-    //上传日志工具类
-    private FileSystemContext fileSystemContext;
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
                           final String workingDir, final Integer logId) {
@@ -79,7 +75,7 @@ public class MythJobProcess implements Runnable{
      */
     @SneakyThrows
     @Override
-    public void run() {
+    public String call() {
         //判断是否执行过
         if (this.isStarted() || this.isComplete()) {
             throw new IllegalStateException("该过程只能使用一次");
@@ -145,8 +141,7 @@ public class MythJobProcess implements Runnable{
             if (this.process.isAlive()){
                 this.process.destroy();
             }
-            logFilePath = outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
-            //TODO 将日志写到fastdfs上
+            return  outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
         }
     }
 
@@ -301,8 +296,4 @@ public class MythJobProcess implements Runnable{
     public String getEffectiveUser() {
         return this.effectiveUser;
     }
-
-    public String getLogContent(){
-        return logFilePath;
-    }
 }

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";
+    }
+}