Skip to content

B
byit-myth-job
Commit 2a6faba5 by huangfusuper
Options

修复重跑BUG

parent 80d1a0f5
Showing with 414 additions and 18 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/ExecuteStatusEnum.java
package com.byit.enums;
/**
* @Description 调度状态
* @Description 执行状态
* @Author guo_m
* @Date 2020-03-31
*/
public enum ExecuteStatusEnum {
SUCCESS(1, "成功"),
FAIL(2, "失败"),
REPAIR_SUCCESS(3, "补批成功"),
REPAIR_FAIL(4, "补批失败"),
KILL(5, "杀死")
RUNING("0","运行中"),
SUCCESS("1", "成功"),
FAIL("2", "失败"),
REPAIR_SUCCESS("3", "补批成功"),
REPAIR_FAIL("4", "补批失败"),
KILL("5", "杀死"),
PARENT_NODE_FAIL("6","上级节点执行失败")
;
private Integer code;
private String code;
private String msg;
private ExecuteStatusEnum(Integer code, String msg){
private ExecuteStatusEnum(String code, String msg){
this.code = code;
this.msg = msg;
}
public Integer getCode(){
public String getCode(){
return this.code;
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/exceptions/SuperiorNodeRunException.java 0 → 100644
package com.byit.exceptions;
/**
* 上级节点执行失败的异常类
* @author huangfu
*/
public class SuperiorNodeRunException extends RuntimeException {
public SuperiorNodeRunException(String message) {
super(message);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/NodeVerification.java 0 → 100644
package com.byit.flowservice;
import com.byit.model.JobTask;
/**
* 做节点验证的数据
* 具体注释看下方方法级的注释
* @author huangfu
*/
public interface NodeVerification {
/**
* 上级节点的状态
* 这个方法主要就是检测当前节点的上级节点的状态
* 当上级节点处于未执行 执行中 失败重试中 时返回的状态为 false 即当前节点不可运行
* 当上级节点全部执行成功时 返回false
* 当上级节点失败重试完毕切结果依旧是失败的情况下 会抛出一个异常信息，即上级节点执行失败
* @param thisJobTask 当前节点
* @return 成功 失败 运行中 未重试完毕
*/
boolean superiorNodeStatus(JobTask thisJobTask);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/impl/NodeVerificationImpl.java 0 → 100644
package com.byit.flowservice.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.exceptions.SuperiorNodeRunException;
import com.byit.flowservice.NodeVerification;
import com.byit.job.exceptions.BusinessException;
import com.byit.model.JobTask;
import com.byit.model.JobTaskRunLog;
import com.byit.service.JobTaskRunLogService;
import com.byit.util.MythStringUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.stream.Collectors;
/**
* 具体的详细注释参见{@link NodeVerification}
* @author huangfu
*/
@Component
public class NodeVerificationImpl implements NodeVerification {
public static final String NODEDEPEND_SPLIT = ",";
/**
* 日志节点操作
*/
private final JobTaskRunLogService jobTaskRunLogService;
public NodeVerificationImpl(JobTaskRunLogService jobTaskRunLogService) {
this.jobTaskRunLogService = jobTaskRunLogService;
}
/**
* 具体这个方法的概述参见{@link NodeVerification#superiorNodeStatus(com.byit.model.JobTask)}
*
* 这个方法判断上级是否执行完毕的逻辑
* 1.查询该节点的依赖节点
* 2.然后返回所有依赖节点的日志信息，判断上级节点是否和查询出来的数目相同，相同就证明上级节点已经全部完成了
* 3.判断上级节点是否全部成功
* 成功：true
* 失败：
* 判断失败的节点是否已经重试完毕
* 完毕：
* 判断该节点是否是弱引用 弱引用直接返回true
* 抛出异常
* 重试中：false
* @param thisJobTask 当前节点
* @return 返回的是该节点是否可以执行
*/
@Override
public boolean superiorNodeStatus(JobTask thisJobTask) {
//获取该节点的运行标识
String runId = thisJobTask.getRunId();
//查询该节点的依赖节点
String nodeDepend = thisJobTask.getNodeDepend();
if(StringUtils.isBlank(nodeDepend)){
//对于这个操作，外部捕获到这个异常后应该将该节点写入日志，并设置异常信息
throw new BusinessException(NodeRunStatusPropertyEnum.NODE_RELY_ERROR.getMsg());
}
String[] split = nodeDepend.split(NODEDEPEND_SPLIT);
//根据依赖节点查询对应的日志信息
List<Integer> nodeDependIntegers = MythStringUtil.stringArrayConvertIntegerArray(split);
//这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(nodeDependIntegers, runId);
//判断上级节点是否和查询出来的数目相同
if (nodeDependIntegers.size() == jobTaskRunLogList.size()){
//如果腹肌节点全部完成 那么判断父级节点是否全部成功
//过滤失败的节点
List<JobTaskRunLog> errorJobLog = jobTaskRunLogList.stream().
filter(jobTaskRunLog -> (
NodeRunStatusPropertyEnum.RUN_FAILURE.getCode().equals(jobTaskRunLog.getRunCode())
|| NodeRunStatusPropertyEnum.RE_RUN_FAILURE.getCode().equals(jobTaskRunLog.getRunCode())
|| NodeRunStatusPropertyEnum.PARENT_NODE_FAILED.getCode().equals(jobTaskRunLog.getRunCode())))
.collect(Collectors.toList());
//当上级节点的失败个数为0时 返回true
if(CollectionUtil.isEmpty(errorJobLog)){
return true;
}else {
//查看失败节点的重试次数是不是为0
for (JobTaskRunLog jobTaskRunLog : errorJobLog) {
if (jobTaskRunLog.getFailedRemainingCount() != null) {
if (jobTaskRunLog.getFailedRemainingCount() > 0){
//上级节点的失败原因还不能是被杀死和上级节点执行失败的，只有这样他才有重试的资格
if (!(NodeRunStatusPropertyEnum.KILL.getCode().equals(jobTaskRunLog.getRunCode()) ||
NodeRunStatusPropertyEnum.PARENT_NODE_FAILED.getCode().equals(jobTaskRunLog.getRunCode()))){
return false;
}
}
}
}
//这里还有一层判断，就是当上级节点执行失败了，而且重试完了，不能立即断定该节点就一定要执行快速失败，因为如果该节点是弱引用
//那么该节点依旧能够执行，这一段逻辑之所以不在首行进行判断是因为，无论他是不是弱引用，他都要等待上级节点执行完毕后才能执行
//判断该节点是否是弱引用
if (NodePropertyEnum.WEAK_NODE.getCode().equals(thisJobTask.getSuperSuccessRun())) {
return true;
}
throw new SuperiorNodeRunException("上级节点执行失败");
}
}
return false;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowServiceImpl.java
......@@ -5,7 +5,6 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.byit.dto.FlowConditionDto;
import com.byit.dto.web.ResponseResult;
import com.byit.enums.ExecuteStatusEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodePropertyEnum;
import com.byit.job.utils.CronExpression;
......@@ -23,7 +22,6 @@ import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestBody;
import javax.annotation.Resource;
import java.util.*;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/RunRecordingAndJobTaskService.java
......@@ -19,6 +19,12 @@ public interface RunRecordingAndJobTaskService {
void saveRunRecordingAndTask(JobTask jobTask) throws Exception;
/**
* 修改运行记录，保存task节点 修改log日志
* @param jobTask
*/
void updateRunRecordingAndSaveTask(JobTask jobTask);
/**
* 修改运行实例 保存任务节点
* @param waitingRecord
*/
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndJobTaskServiceImpl.java
package com.byit.service.mapservice.impl;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.EmailEnum;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.*;
import com.byit.job.utils.DateUtil;
import com.byit.model.*;
import com.byit.service.*;
......@@ -112,7 +109,9 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
.flowNodeCount(virFlow.getFlowNodeCount())
.isAlarm(EmailEnum.IS_ALARM_NO.getCode())
.isInner(FlowPropertyEnum.IS_INNER.getCode())
.failFast(RunRecordingEnum.FAIL_FAST_NO.getCode()).build();
.failFast(RunRecordingEnum.FAIL_FAST_NO.getCode())
.workspaceId(virFlow.getWorkspaceId())
.build();
runRecordingService.saveRunRecording(runRecording);
log.info("-----------【虚节点对应节点保存到任务表】--------------");
//获取所有的节点，开始将所有节点保存到任务表
......@@ -144,8 +143,53 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
}
@Override
public void updateRunRecordingAndSaveTask(JobTask jobTask) {
//节点重跑需要将节点对应的信息从日志里面拉取出来
Integer mapFlowId = jobTask.getMapFlowId();
String reRunId = jobTask.getReRunId();
List<JobTaskRunLogWithBLOBs> reNodeLog = jobTaskRunLogService.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(mapFlowId, reRunId);
for (JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs : reNodeLog) {
jobTaskRunLogWithBLOBs.setRunCode(ExecuteStatusEnum.RUNING.getCode());
jobTaskRunLogWithBLOBs.setOperator(jobTask.getOperator());
jobTaskRunLogWithBLOBs.setScheduleType(jobTask.getScheduleType());
jobTaskRunLogWithBLOBs.setReRunId(jobTask.getReRunId());
jobTaskRunLogWithBLOBs.setReRunId(jobTask.getRunId());
//修改日志
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLogWithBLOBs);
//将日志修改为task节点
JobTask logConvertTask = logConvertTask(jobTaskRunLogWithBLOBs);
logConvertTask.setReRunId(jobTask.getReRunId());
logConvertTask.setOperator(jobTask.getOperator());
logConvertTask.setScheduleType(jobTask.getScheduleType());
jobTaskService.addMythJobTask(logConvertTask);
}
//修改之前的日志为运行中
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(mapFlowId, reRunId);
runRecordingByFlowIdAndRunId.setFlowStatus("2");
runRecordingService.updateRunRecordingById(runRecordingByFlowIdAndRunId);
runRecordingByFlowIdAndRunId.setRunId(jobTask.getRunId());
runRecordingByFlowIdAndRunId.setReRunId(reRunId);
runRecordingByFlowIdAndRunId.setOperator(jobTask.getOperator());
runRecordingByFlowIdAndRunId.setScheduleType(jobTask.getScheduleType());
runRecordingByFlowIdAndRunId.setRecordingId(null);
//新增一条重跑的运行记录
runRecordingService.saveRunRecording(runRecordingByFlowIdAndRunId);
}
private JobTask logConvertTask(JobTaskRunLogWithBLOBs jobTaskRunLogWithBLOBs){
JobTask jobTask = new JobTask();
BeanUtils.copyProperties(jobTaskRunLogWithBLOBs,jobTask);
jobTask.setFailedRetryCount(jobTaskRunLogWithBLOBs.getFailedRemainingCount());
jobTask.setPriority("1");
jobTask.setFailedRetryInterval(1000L);
jobTask.setRunParam(jobTaskRunLogWithBLOBs.getRunParams());
jobTask.setTriggerStatus("1");
return jobTask;
}
@Override
public void updateRunRecordingAndTask(WaitingRecord waitingRecord) {
log.info("---------开始查询等待工作流{}对应的数据-------------",waitingRecord);
log.debug("---------开始查询等待工作流{}对应的数据-------------",waitingRecord);
RunRecording runRecording = runRecordingService.findAllByRunID(waitingRecord.getRunId());
//将运行实例改为以执行
runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_NOT_RUN.getCode());
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/mapservice/impl/RunRecordingAndLogServiceImpl.java
......@@ -60,6 +60,8 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService
Date thisTime = new Date();
log.setStartTime(thisTime);
log.setEndTime(thisTime);
log.setRunId(runId);
log.setFlowName(runRecording.getFlowName());
log.setTriggerTime(thisTime);
jobTaskRunLogService.saveJobTaskRunLog(log);
});
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ClosingExampleThreadRunHelper.java
......@@ -71,6 +71,14 @@ public class ClosingExampleThreadRunHelper extends BaseThreadRunHelper implement
runRecording.setEndTime(new Date());
runRecordingService.updateRunRecordingById(runRecording);
applicationEventPublisher.publishEvent(new EndFlowEvent(this,flowId));
if (StringUtils.isNotBlank(runRecording.getReRunId())) {
String reRunId = runRecording.getReRunId();
Integer flowId1 = runRecording.getFlowId();
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(flowId1, reRunId);
runRecordingByFlowIdAndRunId.setFlowStatus(RunRecordingEnum.FLOW_STATUS_IS_END.getCode());
runRecordingService.updateRunRecordingById(runRecordingByFlowIdAndRunId);
}
}
});
return UNIVERSAL_WAIT_TIME;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/JobTaskThreadRunHelper.java 0 → 100644
package com.byit.thread.helper;
import cn.hutool.core.collection.CollectionUtil;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodeNameEnum;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.exceptions.SuperiorNodeRunException;
import com.byit.flowservice.NodeVerification;
import com.byit.model.JobTask;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.service.JobTaskService;
import com.byit.service.RunRecordingService;
import com.byit.service.mapservice.RunRecordingAndJobTaskService;
import com.byit.service.mapservice.TaskAndLogServer;
import com.byit.service.mapservice.TaskAndScheduleService;
import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
/**
* new jobTask的扫描线程
* @author huangfu
*/
@Slf4j
@Component
public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
/**
* 读取任务节点的预读
*/
private static final long PRE_READ_MS = 7000;
private static final String LOCK_NAME = "job_task_lock";
private final DataSource dataSource;
private final NodeVerification nodeVerification;
private final JobTaskService jobTaskService;
private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
/**
* 任务表和排期表的组合操作
*/
private final TaskAndScheduleService taskAndScheduleService;
/**
* 任务表和日志表操作
*/
private final TaskAndLogServer taskAndLogServer;
/**
* 运行记录表信息操作
*/
private final RunRecordingService runRecordingService;
public JobTaskThreadRunHelper(DataSource dataSource, NodeVerification nodeVerification, JobTaskService jobTaskService, RunRecordingAndJobTaskService runRecordingAndJobTaskService, TaskAndScheduleService taskAndScheduleService, TaskAndLogServer taskAndLogServer, RunRecordingService runRecordingService) {
this.dataSource = dataSource;
this.nodeVerification = nodeVerification;
this.jobTaskService = jobTaskService;
this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
this.taskAndScheduleService = taskAndScheduleService;
this.taskAndLogServer = taskAndLogServer;
this.runRecordingService = runRecordingService;
}
@Override
public Long start() {
long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态，而且七秒内即将运行的任务 而且还不是暂停的节点
List<JobTask> jobTasks = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
//即将执行的节点如果不为null的话
if (CollectionUtil.isNotEmpty(jobTasks)) {
//这个是排期表
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<>(64);
//遍历七秒内要执行的数据
for (JobTask jobTask : jobTasks) {
/**
* 判断节点状态
* 1.虚节点状态
* 验证上级状态，
* 虚节点状态是映射了一个工作流，需要将该节点映射的工作流下所由的几点拉取到任务表
* 2.普通节点也有两种状态：
* I.开始节点：开始节点不需要验证上级工作流，直接放行执行
* II.正常节点：正常节点需要验证上级节点，首先判断自己是否收弱引用，如果是弱引用那么需要判断
* 上级节点是否已经全部都执行完了，执行完后不论成功与否都执行，同时工作流的运行结果
* 只与end节点关联
*/
if(FlowPropertyEnum.IS_INNER.getCode().equals(jobTask.getIsVirtual())) {
try {
if (nodeVerification.superiorNodeStatus(jobTask)) {
log.debug("检测到虚节点：{}，切满足执行条件，开始运行", jobTask);
Integer scheduleType = jobTask.getScheduleType();
if (ScheduleTypeEnum.REPEAT.getCode().equals(scheduleType)){
log.debug("虚节点{},是重跑状态", jobTask);
runRecordingAndJobTaskService.updateRunRecordingAndSaveTask(jobTask);
}else {
runRecordingAndJobTaskService.saveRunRecordingAndTask(jobTask);
}
}
}catch (SuperiorNodeRunException se) {
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
} catch (Exception e) {
e.printStackTrace();
}
} else {
//先处理开始节点
if(NodeNameEnum.START_NODE.getNodeName().equals(jobTask.getNodeName())){
startNodeOperating(jobTask,jobTaskSchedules);
}else {
//普通节点
try {
//断定上级节点是否执行成功
if (nodeVerification.superiorNodeStatus(jobTask)) {
runJobTask(jobTask,jobTaskSchedules);
}
}catch (SuperiorNodeRunException se) {
//删除这个数据 并且添加到日志
taskAndLogServer.addRunLogAndRemoveTask(jobTask);
}
}
}
}
//执行保存到排表 删除任务表操作
taskAndScheduleService.saveScheduleAndDeleteTask(jobTaskSchedules);
}else {
return UNIVERSAL_WAIT_TIME;
}
return PRE_READ_MS;
}
/**
* start节点的操作
* @param thisJobTask 当前的任务节点
* @param jobTaskSchedules 排期集合
*/
private void startNodeOperating(JobTask thisJobTask,List<JobTaskSchedule> jobTaskSchedules){
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(thisJobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
//更改运行记录为运行中
String runId = thisJobTask.getRunId();
Integer flowId = thisJobTask.getFlowId();
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(flowId, runId);
runRecordingByFlowIdAndRunId.setFlowStatus(FlowPropertyEnum.FLOW_RUN_ING.getCode());
runRecordingByFlowIdAndRunId.setStartTime(new Date());
runRecordingService.updateRunRecordingById(runRecordingByFlowIdAndRunId);
}
@Override
public DataSource getDataSource() {
return dataSource;
}
@Override
public String getLockName() {
return LOCK_NAME;
}
/**
* 运行符合条件的人物节点 将task节点转换成排期节点 保存到集合
* @param jobTask 任务节点
* @param jobTaskSchedules 排期集合
*/
private void runJobTask(JobTask jobTask,List<JobTaskSchedule> jobTaskSchedules){
//到这里 父类节点一定是全部都执行成功了,或者是弱节点！
JobTaskSchedule jobTaskSchedule = new JobTaskSchedule();
BeanUtils.copyProperties(jobTask,jobTaskSchedule);
jobTaskSchedules.add(jobTaskSchedule);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java
......@@ -32,7 +32,7 @@ import java.util.stream.Collectors;
* 任务表操作 向排期表添加任务节点并执行
* @author huangfu
*/
@Component
@Deprecated
@Slf4j
public class TaskThreadRunHelper extends BaseThreadRunHelper {
/**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/MythStringUtil.java 0 → 100644
package com.byit.util;
import java.util.ArrayList;
import java.util.List;
/**
* 对特定字符串的封装
* @author huangfu
*/
public class MythStringUtil {
/**
* 将对应字符串数组转换成整形集合
* @param datas 数据集合
* @return 转换数据
*/
public static List<Integer> stringArrayConvertIntegerArray(String[] datas){
List<Integer> convertData = new ArrayList<>(8);
if(datas != null && datas.length>0){
for (String data : datas) {
convertData.add(Integer.parseInt(data));
}
}
return convertData;
}
}
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