Skip to content

B
byit-myth-job
Commit 85d7528a by huangfusuper
Options

修改立即运行的方式

parent 211b530f
Showing with 315 additions and 78 deletions
byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeServiceImpl.java
......@@ -22,6 +22,7 @@ import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.service.ApiNodeService;
import com.byit.service.JobTaskRunLogService;
import com.byit.task.JavaNodeExecutorTask;
import com.byit.task.JavaTaskJobTask;
import com.byit.task.ScriptExecutorJobTask;
import com.byit.utils.ValidationUtil;
......@@ -112,13 +113,7 @@ public class ApiNodeServiceImpl implements ApiNodeService {
stringRedisTemplate.opsForList().rightPush(runKey, JSON.toJSONString(runLog, WriteClassName));
TimerTask timerTask;
if (NodeTypeEnum.JAVA.getCode().equals(runNode.getJobType())) {
//构建调度执行器
JavaTask javaTask = new JavaTask();
javaTask.setTriggerTime(100L);
javaTask.setJobName(schedule.getNodeName());
javaTask.setTaskName(schedule.getNodeName());
javaTask.setParam(schedule.getRunParam());
timerTask = new JavaTaskJobTask(javaTask);
timerTask = new JavaNodeExecutorTask(schedule);
}else{
timerTask = new ScriptExecutorJobTask(schedule);
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/FastRunLogService.java 0 → 100644
package com.byit.service;
import com.byit.model.JobTaskSchedule;
/**
* @author huangfu
*/
public interface FastRunLogService {
/**
* 错误日志快速生成
* @param mythJobTaskSchedule 运势排期
* @param code 错误码
*/
void fastErrorLog(JobTaskSchedule mythJobTaskSchedule, String code);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/FlowStatusService.java 0 → 100644
package com.byit.service;
/**
* @author huangfu
*/
public interface FlowStatusService {
/**
* 探查工作流是否存活
* @param flowId
* @param runId
* @return
*/
String checkFlowStatusIsKill(Integer flowId,String runId);
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FastRunLogServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.enums.EmailEnum;
import com.byit.enums.RunRecordingEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.enums.task.RunTypeEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.service.FastRunLogService;
import com.byit.service.JobTaskRunLogService;
import com.byit.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.util.Date;
/**
* @author huangfu
*/
@Slf4j
@Service
public class FastRunLogServiceImpl implements FastRunLogService {
private final JobTaskRunLogService jobTaskRunLogService;
public static final String LINE = "\n";
public FastRunLogServiceImpl(JobTaskRunLogService jobTaskRunLogService) {
this.jobTaskRunLogService = jobTaskRunLogService;
}
@Override
public void fastErrorLog(JobTaskSchedule mythJobTaskSchedule, String code) {
log.debug("-----------saveErrorLog--保存失败脚本调度日志开始------------");
String msg = "";
if(RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(code)){
msg = RunRecordingEnum.RUN_FLOW_KILL.getMessage();
}
if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)) {
msg = RunRecordingEnum.FAIL_FAST_YES.getMessage();
}
if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)){
code = RunResultEnum.RUN_ERROR.getCode();
}
//获取值放入调度轮之前保存的日志
JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
//放置调度记录 如果执行次数大于一
if (jobTaskRunLogById.getRunCount()>1) {
//上一次调度的日志
String triggerMsg = jobTaskRunLogById.getTriggerMsg();
jobTaskRunLog.setTriggerMsg(triggerMsg+msg+LINE);
}else{
jobTaskRunLog.setTriggerMsg(msg+LINE);
}
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
jobTaskRunLog.setRunType(RunTypeEnum.EXECUTIVE_MACHINE_RUN.getCode());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
//流被kill他的调度肯定失败
jobTaskRunLog.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
//将剩余失败重试次数刷新为0
jobTaskRunLog.setFailedRemainingCount(0);
//将执行状态更改为被杀死
jobTaskRunLog.setRunCode(code);
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+msg+LINE);
}else{
jobTaskRunLog.setRunMsg(msg+ LINE);
}
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
log.info("-----------saveErrorLog--保存KILL脚本调度日志结束------------");
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/FlowStatusServiceImpl.java 0 → 100644
package com.byit.service.impl;
import com.byit.enums.RunRecordingEnum;
import com.byit.model.RunRecording;
import com.byit.service.FlowStatusService;
import com.byit.service.RunRecordingService;
import org.springframework.stereotype.Service;
/**
* 工作流状态的接口
* @author huangfu
*/
@Service
public class FlowStatusServiceImpl implements FlowStatusService {
private final RunRecordingService runRecordingService;
public FlowStatusServiceImpl(RunRecordingService runRecordingService) {
this.runRecordingService = runRecordingService;
}
@Override
public String checkFlowStatusIsKill(Integer flowId, String runId) {
RunRecording runRecordingByFlowIdAndRunId = runRecordingService.findRunRecordingByFlowIdAndRunId(flowId, runId);
if(runRecordingByFlowIdAndRunId != null){
if (RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(runRecordingByFlowIdAndRunId.getFlowRunResult())) {
return RunRecordingEnum.RUN_FLOW_KILL.getCode();
}
if (RunRecordingEnum.FAIL_FAST_YES.getCode().equals(runRecordingByFlowIdAndRunId.getFailFast())) {
return RunRecordingEnum.FAIL_FAST_YES.getCode();
}
}
return null;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaNodeExecutorTask.java 0 → 100644
package com.byit.task;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.plugin.RunLog;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.CommunicationParam;
import com.byit.service.FastRunLogService;
import com.byit.service.FlowStatusService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunJavaServiceImpl;
import com.byit.util.ServiceInfoUtil;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
/**
* java节点执行
* @author huangfu
*/
@Slf4j
public class JavaNodeExecutorTask implements TimerTask {
private static final Integer INIT_SLEEP_TIME = 100;
private final JobTaskSchedule mythJobTaskSchedule;
public JavaNodeExecutorTask(JobTaskSchedule mythJobTaskSchedule) {
this.mythJobTaskSchedule = mythJobTaskSchedule;
}
@Override
public void run(Timeout timeout) throws Exception {
log.debug("-------------开始交验工作流时否正在运行中---------------");
String code = checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(), mythJobTaskSchedule.getRunId());
if(code != null){
log.warn("--------------该工作流已经被杀死快速失败，执行快速失败！-------------------");
saveErrorLog(mythJobTaskSchedule,code);
return;
}
log.debug("-----------------工作流校验完成-------------");
//获取任务级别 1最低 2最高
String priority = mythJobTaskSchedule.getPriority();
if(NodePropertyEnum.ADVANCED_NODE.getCode().equals(priority)){
MythJobAutoConfigure.ADVANCED_JOB_THREAD_POOL.execute(()-> runJob(mythJobTaskSchedule));
}else{
MythJobAutoConfigure.LOW_LEVEL_JOB_THREAD_POOL.execute(()-> runJob(mythJobTaskSchedule));
}
}
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
RunJavaServiceImpl service = SpringUtil.getBean(RunJavaServiceImpl.class);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLogWithBLOBs jobTaskRunLogById = null;
try{
log.info("--------------logid:{}----------",mythJobTaskSchedule.getLogId());
jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
jobTaskRunLogById = spinLock(jobTaskRunLogById);
//构建JAVA
PluginRpcRequestPacket request = new PluginRpcRequestPacket();
String ipAndPort = ServiceInfoUtil.getIpAndPort();
String callbackUrl = "http://" + ipAndPort + "/myth-job-admin/api/callback/callbackRes";
request.setCallbackUrl(callbackUrl);
CommunicationParam communicationParam = new CommunicationParam();
communicationParam.setLogId(jobTaskRunLogById.getLogId()+"");
communicationParam.setCallbackUrl(callbackUrl);
communicationParam.setBody(mythJobTaskSchedule.getRunParam());
request.setParam(communicationParam);
request.setJobName(mythJobTaskSchedule.getNodeName());
jobTaskRunLogById.setLogId(jobTaskRunLogById.getLogId());
PluginRpcResponsePacket pluginRpcResponsePacket = service.runJava(request);
if(pluginRpcResponsePacket.isStatus()){
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_SUCCESS.getCode());
}else{
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
}
jobTaskRunLogById.setTriggerMsg(pluginRpcResponsePacket.getMsg());
jobTaskRunLogById.setJobGroupIp(pluginRpcResponsePacket.getRunIp());
}catch (Exception e){
StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) SpringUtil.getBean("stringRedisTemplate");
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + e.getMessage()).build();
//stringRedisTemplate.convertAndSend("REAL:EXEC:" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
//使用redis 想队尾push一个值
stringRedisTemplate.opsForList().rightPush("REAL:EXEC:" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
log.error("------执行机异常{}-----", e.getMessage());
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
jobTaskRunLogById.setRunCode(RunResultEnum.RUN_ERROR.getCode());
jobTaskRunLogById.setRunMsg(mythJobTaskSchedule.getNodeName()+":"+e.getMessage());
jobTaskRunLogById.setTriggerMsg(mythJobTaskSchedule.getNodeName()+":"+e.getMessage());
}
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLogById);
}
/**
* 模拟自旋锁实现
* @param jobTaskRunLogById
* @return
* @throws InterruptedException
*/
private JobTaskRunLogWithBLOBs spinLock (JobTaskRunLogWithBLOBs jobTaskRunLogById) throws InterruptedException {
log.debug("---------------进入自旋锁状态spinLock---------");
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
int i = 1;
while (jobTaskRunLogById ==null) {
Thread.sleep(i*INIT_SLEEP_TIME);
jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
if(i >= 3){
break;
}
i++;
}
if(jobTaskRunLogById != null){
log.debug("---------------自旋获取锁成功---------");
return jobTaskRunLogById;
}
log.debug("---------------自旋获取锁失败---------");
throw new RuntimeException("服务器分配资源失败");
}
/**
* 保存失败的节点执行信息
* @param mythJobTaskSchedule 节点信息
*/
private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule,String code){
FastRunLogService bean = SpringUtil.getBean(FastRunLogService.class);
bean.fastErrorLog(mythJobTaskSchedule,code);
}
/**
* 校验工作流是否被杀死或者被快速失败
* @param flowId 工作流ID
* @param runId 运行标识
* @return
*/
private String checkFlowStatusIsKill(Integer flowId,String runId){
FlowStatusService bean = SpringUtil.getBean(FlowStatusService.class);
return bean.checkFlowStatusIsKill(flowId,runId);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaTaskJobTask.java
......@@ -47,7 +47,6 @@ public class JavaTaskJobTask implements TimerTask {
PluginRpcRequestPacket request = new PluginRpcRequestPacket();
//保存到日志
Integer logId = saveLog(javaTask);
request.setExtension(logId+"");
String ipAndPort = ServiceInfoUtil.getIpAndPort();
String callbackUrl = "http://" + ipAndPort + "/myth-job-admin/api/callback/callbackRes";
request.setCallbackUrl(callbackUrl);
......@@ -91,7 +90,7 @@ public class JavaTaskJobTask implements TimerTask {
log.setTriggerTime(new Date());
log.setJobType(JAVA_SYNC);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
jobTaskRunLogService.saveJobTaskRunLog(log);
int i = jobTaskRunLogService.saveJobTaskRunLog(log);
return log.getLogId();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -11,6 +11,8 @@ import com.byit.enums.task.RunTypeEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.service.FastRunLogService;
import com.byit.service.FlowStatusService;
import com.byit.service.RunScriptService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
......@@ -47,7 +49,7 @@ public class ScriptExecutorJobTask implements TimerTask {
@Override
public void run(Timeout timeout) {
log.debug("---------开始交验工作流时否正在运行中------------");
log.debug("-----------开始交验工作流时否正在运行中--------------");
String code = checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(), mythJobTaskSchedule.getRunId());
if(code != null){
log.warn("--------------该工作流已经被杀死快速失败，执行快速失败！-------------------");
......@@ -74,17 +76,8 @@ public class ScriptExecutorJobTask implements TimerTask {
* @return
*/
private String checkFlowStatusIsKill(Integer flowId,String runId){
RunRecordingServiceImpl bean = SpringUtil.getBean(RunRecordingServiceImpl.class);
RunRecording runRecordingByFlowIdAndRunId = bean.findRunRecordingByFlowIdAndRunId(flowId, runId);
if(runRecordingByFlowIdAndRunId != null){
if (RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(runRecordingByFlowIdAndRunId.getFlowRunResult())) {
return RunRecordingEnum.RUN_FLOW_KILL.getCode();
}
if (RunRecordingEnum.FAIL_FAST_YES.getCode().equals(runRecordingByFlowIdAndRunId.getFailFast())) {
return RunRecordingEnum.FAIL_FAST_YES.getCode();
}
}
return null;
FlowStatusService bean = SpringUtil.getBean(FlowStatusService.class);
return bean.checkFlowStatusIsKill(flowId,runId);
}
/**
......@@ -204,56 +197,7 @@ public class ScriptExecutorJobTask implements TimerTask {
* @param mythJobTaskSchedule 节点信息
*/
private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule,String code){
log.debug("-----------saveErrorLog--保存KILL脚本调度日志开始------------");
String msg = "";
if(RunRecordingEnum.RUN_FLOW_KILL.getCode().equals(code)){
msg = RunRecordingEnum.RUN_FLOW_KILL.getMessage();
}
if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)) {
msg = RunRecordingEnum.FAIL_FAST_YES.getMessage();
}
if(RunRecordingEnum.FAIL_FAST_YES.getCode().equals(code)){
code = RunResultEnum.RUN_ERROR.getCode();
}
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
//获取值放入调度轮之前保存的日志
JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
//放置调度记录 如果执行次数大于一
if (jobTaskRunLogById.getRunCount()>1) {
//上一次调度的日志
String triggerMsg = jobTaskRunLogById.getTriggerMsg();
jobTaskRunLog.setTriggerMsg(triggerMsg+msg+LINE);
}else{
jobTaskRunLog.setTriggerMsg(msg+LINE);
}
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
jobTaskRunLog.setRunType(RunTypeEnum.EXECUTIVE_MACHINE_RUN.getCode());
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
//流被kill他的调度肯定失败
jobTaskRunLog.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
//将剩余失败重试次数刷新为0
jobTaskRunLog.setFailedRemainingCount(0);
//将执行状态更改为被杀死
jobTaskRunLog.setRunCode(code);
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+msg+LINE);
}else{
jobTaskRunLog.setRunMsg(msg+ LINE);
}
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
log.info("-----------saveErrorLog--保存KILL脚本调度日志结束------------");
FastRunLogService bean = SpringUtil.getBean(FastRunLogService.class);
bean.fastErrorLog(mythJobTaskSchedule,code);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
......@@ -13,6 +13,7 @@ import com.byit.model.JobTaskSchedule;
import com.byit.service.JobTaskScheduleService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.task.JavaBeanJobTask;
import com.byit.task.JavaNodeExecutorTask;
import com.byit.task.JavaTaskJobTask;
import com.byit.task.ScriptExecutorJobTask;
import com.byit.thread.BaseThreadRunHelper;
......@@ -71,12 +72,7 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
if (NodeTypeEnum.JAVA.getType().equals(mythJobTaskSchedule.getJobType())) {
log.debug("------节点{}，开始构建java执行器-------",mythJobTaskSchedule);
//构建调度执行器
JavaTask javaTask = new JavaTask();
javaTask.setTriggerTime(mythJobTaskSchedule.getTriggerTime());
javaTask.setJobName(mythJobTaskSchedule.getNodeName());
javaTask.setTaskName(mythJobTaskSchedule.getNodeName());
javaTask.setParam(mythJobTaskSchedule.getRunParam());
timerTask = new JavaTaskJobTask(javaTask);
timerTask = new JavaNodeExecutorTask(mythJobTaskSchedule);
}else if(NodeTypeEnum.PYTHON.getCode().equals(mythJobTaskSchedule.getJobType())){
log.debug("------节点{}，开始构建脚本执行器-------",mythJobTaskSchedule);
//构建脚本调度执行器
......
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