调度中心添加杀死工作流和杀死任务的API接口

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

@@ -3,6 +3,7 @@ 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 +26,9 @@ public class ApiFlowController {
     @Resource
     private ApiFlowService apiFlowService;
 
+    @Resource
+    private FlowService flowService;
+
     @PostMapping("publish")
     @ApiOperation("发布工作流，并开始调度")
     public String publishFlow(String param) throws Exception {
@@ -52,11 +56,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/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,6 +351,34 @@ 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";

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