Skip to content

B
byit-myth-job
Commit 70f4cad9 by guominglei
Options

杀死工作流调整

parent d3ceff68
Showing with 89 additions and 31 deletions
byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java
package com.byit.api;
import com.byit.dto.executor.KillDto;
import com.byit.dto.plugin.CollectData;
import com.byit.dto.web.ResponseResult;
import com.byit.model.RunRecording;
......@@ -73,14 +74,14 @@ public class ApiFlowController {
@PostMapping("killJob")
@ApiOperation("杀死节点")
public ResponseResult killJob(String param) throws InterruptedException {
Boolean result = flowService.killJob(param);
KillDto result = flowService.killJob(param);
return ResponseResult.ok(result);
}
@PostMapping("killFlow")
@ApiOperation("杀死工作流")
public ResponseResult killFlow(String param) throws InterruptedException {
Boolean result = flowService.killFlow(param);
KillDto result = flowService.killFlow(param);
return ResponseResult.ok(result);
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/FlowService.java
package com.byit.service;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.executor.KillDto;
import com.byit.model.Flow;
import com.byit.model.vo.FlowImportantAllVo;
import com.byit.model.vo.FlowViewVo;
......@@ -113,12 +114,12 @@ public interface FlowService {
* @param param
* @return
*/
Boolean killJob(String param) throws InterruptedException;
KillDto killJob(String param) throws InterruptedException;
/**
* 杀死一个工作流
* @param param
* @return
*/
Boolean killFlow(String param) throws InterruptedException;
KillDto killFlow(String param) throws InterruptedException;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowServiceImpl.java
......@@ -5,11 +5,9 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.StatisticsConditionDto;
import com.byit.dto.executor.KillDto;
import com.byit.dto.web.ResponseResult;
import com.byit.enums.EmailEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.JobTriggerStatusEnums;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.*;
import com.byit.job.utils.CronExpression;
import com.byit.mapper.*;
import com.byit.model.*;
......@@ -360,7 +358,7 @@ public class FlowServiceImpl implements FlowService {
}
@Override
public Boolean killJob(String param) throws InterruptedException {
public KillDto killJob(String param) throws InterruptedException {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonpObject = JSON.parseObject(param);
String runId = jsonpObject.getString("runId");
......@@ -370,6 +368,7 @@ public class FlowServiceImpl implements FlowService {
String nodeName = jsonpObject.getString("nodeName");
ValidationUtil.dataNotBank(nodeName, "节点名称不允许为空！");
Boolean result = true;
StringBuffer errorMsg = new StringBuffer();
JobTaskRunLogWithBLOBs jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
ValidationUtil.dataNotNull(jobTaskRunLog, "不存在【" + flowName + "】下节点【" + nodeName + "】的运行日志,或尚未开始调度!");
ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
......@@ -383,30 +382,43 @@ public class FlowServiceImpl implements FlowService {
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
innerJobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
ValidationUtil.isTrueValidation(!"0".equals(innerJobTaskRunLog.getRunCode()), "该任务已经运行结束！");
ValidationUtil.dataNotBank(innerJobTaskRunLog.getJobGroupIp(), "尚未分配执行机请稍后再试");
if(!"0".equals(jobTaskRunLog.getRunCode())){
log.info("该任务已经运行结束,无需停止！");
continue;
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
result = false;
continue;
}
}
if (!killJob(innerJobTaskRunLog.getLogId(), innerJobTaskRunLog.getJobGroupIp())){
if (!killJob(innerJobTaskRunLog.getLogId(), innerJobTaskRunLog.getJobGroupIp(), innerJobTaskRunLog.getNodeName(), errorMsg)){
result = false;
}
}
}
}else {
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
if(!"0".equals(jobTaskRunLog.getRunCode())){
log.info("该任务已经运行结束,无需停止！");
return new KillDto(result, errorMsg.toString());
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
return new KillDto(false, errorMsg.toString());
}
}
result = killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp(), jobTaskRunLog.getNodeName(), errorMsg);
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
log.warn("已经调度成功但是还未返回具体的调用机器的ip地址");
Thread.sleep(3000);
jobTaskRunLog = jobTaskRunLogMapper.findByRunIdAndFlowAndNode(runId, flowName, nodeName);
ValidationUtil.isTrueValidation(!"0".equals(jobTaskRunLog.getRunCode()), "该任务已经运行结束！");
ValidationUtil.dataNotBank(jobTaskRunLog.getJobGroupIp(), "尚未分配执行机请稍后再试");
}
result = killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp());
return result;
return new KillDto(result, errorMsg.toString());
}
@Override
public Boolean killFlow(String param) throws InterruptedException {
public KillDto killFlow(String param) throws InterruptedException {
ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
JSONObject jsonObject = JSON.parseObject(param);
String runId = jsonObject.getString("runId");
......@@ -446,18 +458,31 @@ public class FlowServiceImpl implements FlowService {
runRecordingMapper.updateRunRecordingById(runRecording);
});
//杀死所有的任务
StringBuffer errorMsg = new StringBuffer();
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogMapper.findbyRunIdAndFlowIdList(runId, flowIdList);
if (null != jobTaskRunLogList && jobTaskRunLogList.size() > 0){
for (JobTaskRunLog jobTaskRunLog : jobTaskRunLogList){
if ("0".equals(jobTaskRunLog.getRunCode()) && !NodePropertyEnum.IS_VIRTUAL.getCode().equals(jobTaskRunLog.getIsVirtual())){
if ("0".equals(jobTaskRunLog.getRunCode()) && ! NodePropertyEnum.IS_VIRTUAL.getCode().equals(jobTaskRunLog.getIsVirtual())){
if (NodeTypeEnum.JAVA.getCode().equals(jobTaskRunLog.getJobType())){
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点类型为JAVA，无法杀死！\n");
result = false;
continue;
}
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(!"0".equals(jobTaskRunLog.getRunCode())){
log.info("该任务已经运行结束,无需停止！");
continue;
}
if (StringUtils.isEmpty(jobTaskRunLog.getJobGroupIp())){
errorMsg.append("【").append(jobTaskRunLog.getNodeName()).append("】节点尚未分配执行机请稍后再试\n");
result = false;
continue;
}
}
if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp())){
if (! killJob(jobTaskRunLog.getLogId(), jobTaskRunLog.getJobGroupIp(), jobTaskRunLog.getNodeName(), errorMsg)){
result = false;
}
}
......@@ -475,10 +500,11 @@ public class FlowServiceImpl implements FlowService {
});
flow.setScanMark("1");
flowMapper.updateByIdSelective(flow);
return result;
return new KillDto(result, errorMsg.toString());
}
private synchronized Boolean killJob(Integer logId, String exectUrl){
private synchronized Boolean killJob(Integer logId, String exectUrl, String nodeName, StringBuffer errorMsg){
//具体访问的URL
String killUrl = "http://" + exectUrl + "/myth-executor-server/processManager/killJob";
Map<String, Object> requestMap = new HashMap<>(2);
......@@ -486,8 +512,10 @@ public class FlowServiceImpl implements FlowService {
String killResult = HttpUtil.post(killUrl, requestMap);
log.info("请求结果{}", killResult);
ResponseResult responseResult = JSON.parseObject(killResult, ResponseResult.class);
ValidationUtil.isTrueValidation(!"SUCCESS".equals(responseResult.getResult()), "杀死job失败，错误信息为：" + responseResult.getMsg());
if (!"SUCCESS".equals(responseResult.getResult())){
errorMsg.append("【").append(nodeName).append("】杀死失败，错误信息为:").append(responseResult.getMsg());
return false;
}
return true;
}
......
byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/executor/KillDto.java 0 → 100644
package com.byit.dto.executor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* @author gml
* @description: 杀死结果
* @date 2020/8/19 17:00
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class KillDto {
/**
* 杀死结果
*/
private Boolean result;
/**
* 详细信息
*/
private String msg;
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment