Skip to content

B
byit-myth-job
Commit 5efb1cd8 by huangfusuper
Options

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

parents 153403b6 e540db31
Showing with 115 additions and 37 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-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java
......@@ -3,7 +3,6 @@ package com.byit.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.ConsistentHash;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.StatisticsConditionDto;
......@@ -26,7 +25,6 @@ import com.byit.util.IDGenerationStrategy;
import com.byit.util.lock.RedissLockUtil;
import com.byit.utils.ValidationUtil;
import com.google.common.base.Joiner;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
......@@ -38,7 +36,7 @@ import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
......@@ -1173,9 +1171,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
@Override
public Boolean stopScheduleByRunId(String runId) {
RunRecording recording = runRecordingMapper.findAllByRunID(runId);
ValidationUtil.dataNotNull(recording , "查无此运行实例！");
ValidationUtil.isTrueValidation("3".equals(recording.getFlowStatus()) || "4".equals(recording.getFlowStatus()) , "该运行实例没有在运行中！");
List<RunRecording> recordingList = runRecordingMapper.findByRunID(runId);
ValidationUtil.dataNotNull(recordingList , "查无此运行实例！");
AtomicReference<Boolean> result = new AtomicReference<>(true);
recordingList.forEach(runRecording -> {
if (("3".equals(runRecording.getFlowStatus()) || "4".equals(runRecording.getFlowStatus())) //如果不是正在运行
&& FlowPropertyEnum.ISNOT_INNER.getCode().equals(runRecording.getIsInner())){ //并且不是内嵌工作流
result.set(false);
}
});
ValidationUtil.isTrueValidation(!result.get(), "该运行实例没有在运行中！");
//暂停工作流调度
runRecordingMapper.stopByRunId(runId);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/RunRecordingMapper.java
......@@ -54,6 +54,13 @@ public interface RunRecordingMapper {
* @return
*/
RunRecording findAllByRunID(String runId);
/**
*根据运行标识查询对应的运行实例
* @param runId
* @return
*/
List<RunRecording> findByRunID(String runId);
/**
* 查询运行中的运行实例
* @return
......
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-admin-core/src/main/resources/mapper/RunRecordingMapper.xml
......@@ -105,6 +105,14 @@
from run_recording
where run_id=#{runId,jdbcType=VARCHAR} and schedule_type != 4
</select>
<select id="findByRunID" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
from run_recording
where run_id=#{runId,jdbcType=VARCHAR} and schedule_type != 4
</select>
<!--查询运行中的-->
<select id="findRunningRunRecording" resultMap="BaseResultMap">
select
......
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