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

byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java

 package com.byit.api;
 
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import com.byit.service.ApiFlowService;
+import com.byit.service.FlowService;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -25,6 +24,9 @@ public class ApiFlowController {
     @Resource
     private ApiFlowService apiFlowService;
 
+    @Resource
+    private FlowService flowService;
+
     @PostMapping("publish")
     @ApiOperation("发布工作流，并开始调度")
     public String publishFlow(String param) throws Exception {
@@ -52,11 +54,19 @@ public class ApiFlowController {
         apiFlowService.repealSchedule(param);
         return "SUCCESS";
     }
-    @PostMapping("killSchedule")
-    @ApiOperation("杀死本次调度")
-    public String killSchedule(String param){
-        apiFlowService.killSchedule(param);
-        return "SUCCESS";
+
+    @PostMapping("killJob")
+    @ApiOperation("杀死节点")
+    public Boolean killJob(String param) throws InterruptedException {
+        Boolean result = flowService.killJob(param);
+        return result;
+    }
+
+    @PostMapping("killFlow")
+    @ApiOperation("杀死工作流")
+    public Boolean killFlow(String runId) throws InterruptedException {
+        Boolean result = flowService.killFlow(runId);
+        return result;
     }
 
     @PostMapping("stopSchedule")

byit-myth-core/myth-admin-core/pom.xml

@@ -59,11 +59,6 @@
             <groupId>io.springfox</groupId>
             <artifactId>springfox-swagger-ui</artifactId>
         </dependency>
-        <dependency>
-            <groupId>com.byit</groupId>
-            <artifactId>byit-validation-starter</artifactId>
-            <version>1.0-SNAPSHOT</version>
-        </dependency>
 
         <dependency>
             <groupId>myth-job</groupId>

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

@@ -103,4 +103,11 @@ public interface JobTaskRunLogMapper {
      * @return
      */
     List<JobTaskRunLog> findByRunIdAndFlowName(@Param("runId")String runId, @Param("flowName")String flowName);
+
+    /**
+     * 根据runid获取所有的运行任务
+     * @param runId
+     * @return
+     */
+    List<JobTaskRunLog> findByRunId(String runId);
 }
\ No newline at end of file

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

@@ -101,7 +101,7 @@ public interface RunRecordingMapper {
      * @param runId
      * @return
      */
-    int killFlow(String runId);
+    int killUnFinshFlow(String runId);
 
     /**
      * 杀死工作流内的某个内嵌工作流
@@ -111,4 +111,10 @@ public interface RunRecordingMapper {
      */
     int killInnerFlow(String runId, String flowName);
 
+    /**
+     * 根据运行id查找一系列运行实例
+     * @param runId
+     * @return
+     */
+    List<RunRecording> findUnFinishByRunId(String runId);
 }
\ No newline at end of file

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

@@ -86,10 +86,15 @@ public interface FlowService {
 
     /**
      * 杀死一条任务
-     * @param runId     运行实例id
-     * @param flowName  工作流名称
-     * @param nodeName  任务名称
+     * @param param
      * @return
      */
-    Boolean killJob(String runId, String flowName, String nodeName) throws InterruptedException;
+    Boolean killJob(String param) throws InterruptedException;
+
+    /**
+     * 杀死一个工作流
+     * @param runId
+     * @return
+     */
+    Boolean killFlow(String runId) throws InterruptedException;
 }

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

@@ -2,6 +2,8 @@ package com.byit.service.impl;
 
 import cn.hutool.http.HttpUtil;
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.JSONPObject;
 import com.byit.enums.FlowPropertyEnum;
 import com.byit.enums.NodePropertyEnum;
 import com.byit.job.utils.CronExpression;
@@ -10,6 +12,7 @@ import com.byit.mapper.*;
 import com.byit.model.Flow;
 import com.byit.model.JobTaskRunLog;
 import com.byit.model.Node;
+import com.byit.model.RunRecording;
 import com.byit.model.vo.FlowVo;
 import com.byit.model.vo.NodeVo;
 import com.byit.service.FlowService;
@@ -303,9 +306,14 @@ public class FlowServiceImpl implements FlowService {
     }
 
     @Override
-    public Boolean killJob(String runId, String flowName, String nodeName) throws InterruptedException {
+    public Boolean killJob(String param) throws InterruptedException {
+        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
+        JSONObject jsonpObject = JSON.parseObject(param);
+        String runId = jsonpObject.getString("runId");
         ValidationUtil.dataNotBank(runId, "运行实例id不允许为空！");
+        String flowName = jsonpObject.getString("flowName");
         ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
+        String nodeName = jsonpObject.getString("nodeName");
         ValidationUtil.dataNotBank(nodeName, "节点名称不允许为空！");
         Boolean result = true;
         JobTaskRunLog jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
@@ -343,12 +351,40 @@ public class FlowServiceImpl implements FlowService {
         return result;
     }
 
+    @Override
+    public Boolean killFlow(String runId) throws InterruptedException {
+        ValidationUtil.dataNotBank(runId, "运行实例不允许3为空！");
+        //杀死运行实例
+        List<RunRecording> runRecordings = runRecordingMapper.findUnFinishByRunId(runId);
+        ValidationUtil.dataNotNull(runRecordings, "该运行id未找到或已经运行完成无需杀死！");
+        Boolean result = true;
+        //杀死未完成的运行实例
+        runRecordingMapper.killUnFinshFlow(runId);
+        //杀死所有的任务
+        List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogMapper.findByRunId(runId);
+        for (JobTaskRunLog jobTaskRunLog : jobTaskRunLogList){
+            if (!"0".equals(jobTaskRunLog.getRunCode())){
+                if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
+                    log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
+                    Thread.sleep(3000);
+                    jobTaskRunLog = jobTaskRunLogMapper.findJobTaskRunLogByLogId(jobTaskRunLog.getLogId());
+                    ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
+                    ValidationUtil.dataNotBank(jobTaskRunLog.getJobGroupIp(), "尚未分配执行机请稍后再试");
+                }
+                if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp())){
+                    result = false;
+                }
+            }
+        }
+        return result;
+    }
+
     private synchronized Boolean killJob(Integer logId, String exectUrl){
         //具体访问的URL
         String killUrl = exectUrl + "/processManager/killJob";
-        Map<String, Integer> requestMap = new HashMap<>(2);
+        Map<String, Object> requestMap = new HashMap<>(2);
         requestMap.put("logId", logId);
-        String killResult = HttpUtil.post(killUrl, JSON.toJSONString(requestMap));
+        String killResult = HttpUtil.post(killUrl, requestMap);
         log.info("请求结果{}", killResult);
         ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
         if ("SUCCESS".equals(responseResult.getResult())){
@@ -356,4 +392,5 @@ public class FlowServiceImpl implements FlowService {
         }
         return false;
     }
+
 }
\ No newline at end of file

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

@@ -84,6 +84,12 @@
     where node_id = #{nodeId} and flow_name = #{flowName}
   </select>
 
+  <select id="findByRunId" resultMap="BaseResultMap">
+    select <include refid="Base_Column_List"/>
+    from job_task_run_log
+    where node_id = #{nodeId}
+  </select>
+
   <select id="findJobTaskRunLogWithBLOBsByFlowIdAndRunId" resultMap="ResultMapWithBLOBs">
     select
     <include refid="Base_Column_List" />

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

@@ -74,6 +74,14 @@
         and flow_status = '3'
   </select>
 
+  <select id="findUnFinishByRunId" parameterType="string" resultMap="BaseResultMap">
+    select
+    <include refid="Base_Column_List" />
+    from run_recording
+    where run_id = #{runId,jdbcType=VARCHAR}
+    and flow_status != '4'
+  </select>
+
   <delete id="deleteById" parameterType="java.lang.Integer">
     <!-- generated @mbg.generated date: 2019-12-25 -->
     delete from run_recording
@@ -341,14 +349,16 @@
             and flow_status = '3'
   </update>
 
-  <update id="killFlow" parameterType="string">
+  <update id="killUnFinshFlow" parameterType="string">
     update run_recording set flow_status = '4',flow_run_result = '5'
      where run_id = #{runId,jdbcType=VARCHAR}
+        and flow_status != '4'
   </update>
 
   <update id="killInnerFlow" >
     update run_recording set flow_status = '4',flow_run_result = '5'
-     where run_id = #{runId,jdbcType=VARCHAR} and flow_name = ${flowName,jdbcType=VARCHAR}
+     where run_id = #{runId,jdbcType=VARCHAR}
+        and flow_name = ${flowName,jdbcType=VARCHAR}
   </update>
 
 </mapper>
\ No newline at end of file

byit-myth-core/myth-core-common/pom.xml

@@ -45,6 +45,16 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.byit</groupId>
+            <artifactId>byit-validation-starter</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webmvc</artifactId>
+        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

byit-myth-admin/src/main/java/com/byit/advice/ExceptionHandle.java → byit-myth-core/myth-core-common/src/main/java/com/byit/job/advice/ExceptionHandle.java

-package com.byit.advice;
+package com.byit.job.advice;
 
 import com.byit.exception.DataValidationException;
 import com.byit.job.enums.IEnum;

byit-myth-admin/src/main/java/com/byit/advice/ResponseAdvice.java → byit-myth-core/myth-core-common/src/main/java/com/byit/job/advice/ResponseAdvice.java

-package com.byit.advice;
+package com.byit.job.advice;
 
 import com.alibaba.fastjson.JSON;
 import com.byit.job.vo.ResponseResult;

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/Test2.java → byit-myth-core/myth-executor-core/src/test/java/com/test/Test2.java

-package com.byit.executor;
+package com.test;
 
 import java.io.BufferedReader;
 import java.io.IOException;

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

 package com.byit;
 
+import com.byit.rpc.remoting.provider.annotation.RpcService;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@@ -10,6 +11,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
  * @date: 2019/11/14 16:03
  **/
 @SpringBootApplication
+@RpcService(http_type = true)
 public class ExecutorServiceApplication {
     public static void main(String[] args) {
         SpringApplication.run(ExecutorServiceApplication.class,args);

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

@@ -4,7 +4,7 @@ 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.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 /**
@@ -12,11 +12,11 @@ import org.springframework.web.bind.annotation.RestController;
  * @Author guo_m
  * @Date 2020-02-22
  */
-@RestController("processManager")
+@RestController
+@RequestMapping("processManager")
 public class ProcessManagerController {
 
     @PostMapping("killJob")
-    @ResponseBody
     public String killJob(Integer logId){
         ValidationUtil.dataNotNull(logId, "日志id不允许为空！");
         MythJobProcess mythJobProcess = JobContentUtil.getJobThread(logId);