Skip to content

B
byit-myth-job
Commit a727c5e9 by huangfusuper
Options

重构补批工作流

parent 0b7fac9d
Showing with 57 additions and 46 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -12,6 +12,7 @@ import com.byit.enums.*;
import com.byit.enums.task.RunResultEnum;
import com.byit.enums.task.RunTypeEnum;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.MythLogUtils;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
......@@ -19,7 +20,6 @@ import com.byit.service.FastRunLogService;
import com.byit.service.FlowStatusService;
import com.byit.service.RunScriptService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.job.utils.MythLogUtils;
import com.byit.util.GetRegConfig;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
......@@ -28,13 +28,17 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.*;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
import java.util.stream.Collectors;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
/**
* 脚本的执行调用器
*
* @author huangfu
*/
@Slf4j
......@@ -56,9 +60,9 @@ public class ScriptExecutorJobTask implements TimerTask {
public void run(Timeout timeout) {
log.debug("-----------开始交验工作流时否正在运行中--------------");
String code = checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(), mythJobTaskSchedule.getRunId());
if(code != null){
if (code != null) {
log.warn("--------------该工作流已经被杀死快速失败，执行快速失败！-------------------");
saveErrorLog(mythJobTaskSchedule,code);
saveErrorLog(mythJobTaskSchedule, code);
return;
}
log.debug("-----------------工作流校验完成-------------");
......@@ -66,26 +70,27 @@ public class ScriptExecutorJobTask implements TimerTask {
//获取任务级别 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));
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));
}
}
/**
* 校验工作流是否被杀死或者被快速失败
*
* @param flowId 工作流ID
* @param runId 运行标识
* @return
*/
private String checkFlowStatusIsKill(Integer flowId,String runId){
private String checkFlowStatusIsKill(Integer flowId, String runId) {
FlowStatusService bean = SpringUtil.getBean(FlowStatusService.class);
return bean.checkFlowStatusIsKill(flowId,runId);
return bean.checkFlowStatusIsKill(flowId, runId);
}
private void paramBuild(JobTaskSchedule mythJobTaskSchedule){
private void paramBuild(JobTaskSchedule mythJobTaskSchedule) {
//获取运行时参数
String runParam = mythJobTaskSchedule.getRunParam();
//获取命令
......@@ -93,7 +98,7 @@ public class ScriptExecutorJobTask implements TimerTask {
//转换参数对象为参数包装体
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
//当参数包装体不为空时 证明存在参数 或私有或公有
if(runParamWrapped != null) {
if (runParamWrapped != null) {
//获取到私有参数
String privateParam = runParamWrapped.getPrivateParam();
//替换运行参数中的时间参数
......@@ -101,14 +106,14 @@ public class ScriptExecutorJobTask implements TimerTask {
|| ScheduleTypeEnum.REAL.getCode().equals(mythJobTaskSchedule.getScheduleType())) {
privateParam = PlaceholderUtils.formatBizDateParam(privateParam, PlaceholderEnum.DATE_PLACEHOLDER.getName(), 1);
if(StringUtils.isNoneBlank(privateParam)) {
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam,ScriptParamAndPlaceholderDto.class);
if (StringUtils.isNoneBlank(privateParam)) {
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam, ScriptParamAndPlaceholderDto.class);
Map<String, String> placeholder = scriptParamAndPlaceholderDto.getPlaceholder();
if(CollectionUtil.isNotEmpty(placeholder)) {
if(placeholder.containsKey(PlaceholderEnum.NOW_DATE_PLACEHOLDER.getName())){
if (CollectionUtil.isNotEmpty(placeholder)) {
if (placeholder.containsKey(PlaceholderEnum.NOW_DATE_PLACEHOLDER.getName())) {
String nowDateFormatName = placeholder.get(PlaceholderEnum.NOW_DATE_PLACEHOLDER.getName());
String dateFormat = DateUtil.dateFormat(new Date(), nowDateFormatName);
privateParam = PlaceholderUtils.formatBizDateParam(privateParam, PlaceholderEnum.NOW_DATE_PLACEHOLDER.getName(), dateFormat,0);
privateParam = PlaceholderUtils.formatBizDateParam(privateParam, PlaceholderEnum.NOW_DATE_PLACEHOLDER.getName(), dateFormat, 0);
}
}
......@@ -118,22 +123,22 @@ public class ScriptExecutorJobTask implements TimerTask {
}
//获取到公有参数
Map<String,String> publicParam = runParamWrapped.getPublicParamMap();
Map<String, String> publicParam = runParamWrapped.getPublicParamMap();
//将私有参数转换为对应的参数DTO
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = null;
if(StringUtils.isNoneBlank(privateParam)){
scriptParamAndPlaceholderDto = JSON.parseObject(privateParam,ScriptParamAndPlaceholderDto.class);
if (StringUtils.isNoneBlank(privateParam)) {
scriptParamAndPlaceholderDto = JSON.parseObject(privateParam, ScriptParamAndPlaceholderDto.class);
}
//脚本参数不为空的时候
if(scriptParamAndPlaceholderDto != null) {
if (scriptParamAndPlaceholderDto != null) {
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
if(CollectionUtil.isNotEmpty(param)){
command = PlaceholderUtils.commandReplace(command,param);
if (CollectionUtil.isNotEmpty(param)) {
command = PlaceholderUtils.commandReplace(command, param);
}
}
//公共参数不为空的时候
if(CollectionUtil.isNotEmpty(publicParam)) {
command = PlaceholderUtils.commandReplace(command,publicParam );
if (CollectionUtil.isNotEmpty(publicParam)) {
command = PlaceholderUtils.commandReplace(command, publicParam);
}
mythJobTaskSchedule.setRunCommand(command);
}
......@@ -141,16 +146,17 @@ public class ScriptExecutorJobTask implements TimerTask {
/**
* 运行任务
*
* @param mythJobTaskSchedule 运行的排期表
*/
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
try{
try {
paramBuild(mythJobTaskSchedule);
mythJobTaskSchedule.setRunCommand(mythJobTaskSchedule.getRunCommand());
RunScriptService runScriptService = SpringUtil.getBean(RunScriptService.class);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
log.info("--------------logid:{}----------",mythJobTaskSchedule.getLogId());
log.info("--------------logid:{}----------", mythJobTaskSchedule.getLogId());
JobTaskRunLogWithBLOBs jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
jobTaskRunLogById = spinLock(jobTaskRunLogById);
ScriptDto scriptDto = new ScriptDto();
......@@ -161,16 +167,19 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
GetRegConfig getRegConfig = SpringUtil.getBean(GetRegConfig.class);
TreeSet<String> gatewayConf = getRegConfig.getGatewayConf();
if(CollectionUtil.isEmpty(gatewayConf)) {
throw new RuntimeException("网关地址为空！");
}
List<String> callUrlList = gatewayConf.stream().map(callbackIp -> String.format("%s%s%s", HTTP_PRE, callbackIp, HTTP_SUFFIX)).collect(Collectors.toList());
scriptDto.setCallbackUrl(JSON.toJSONString(callUrlList));
//二次执行的情况下 会有这个信息
scriptDto.setLogRemotePath(jobTaskRunLogById.getLogRemotelyPath());
dispatchResponseDto = runScriptService.runScript(scriptDto);
}catch (Exception e){
} catch (Exception e) {
if (mythJobTaskSchedule.getScheduleType().equals(4)) {
StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) SpringUtil.getBean("stringRedisTemplate");
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + MythLogUtils.getMessage(e)).build();
stringRedisTemplate.opsForList().rightPush("REAL:EXEC:" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
stringRedisTemplate.opsForList().rightPush("REAL:EXEC:" + mythJobTaskSchedule.getLogId(), JSON.toJSONString(runLog, WriteClassName));
}
log.error("------执行机异常{}-----", MythLogUtils.getMessage(e));
......@@ -178,30 +187,31 @@ public class ScriptExecutorJobTask implements TimerTask {
dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getCode());
}
saveLog(mythJobTaskSchedule,dispatchResponseDto);
saveLog(mythJobTaskSchedule, dispatchResponseDto);
}
/**
* 模拟自旋锁实现
*
* @param jobTaskRunLogById
* @return
* @throws InterruptedException
*/
private JobTaskRunLogWithBLOBs spinLock (JobTaskRunLogWithBLOBs jobTaskRunLogById) 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);
while (jobTaskRunLogById == null) {
Thread.sleep(i * INIT_SLEEP_TIME);
jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
if(i >= 3){
if (i >= 3) {
break;
}
i++;
}
if(jobTaskRunLogById != null){
if (jobTaskRunLogById != null) {
log.debug("---------------自旋获取锁成功---------");
return jobTaskRunLogById;
}
......@@ -209,7 +219,7 @@ public class ScriptExecutorJobTask implements TimerTask {
throw new RuntimeException("服务器分配资源失败");
}
private void saveLog(JobTaskSchedule mythJobTaskSchedule, DispatchResponseDto dispatchResponseDto){
private void saveLog(JobTaskSchedule mythJobTaskSchedule, DispatchResponseDto dispatchResponseDto) {
log.debug("-----------saveLog--保存脚本调度日志开始------------");
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
......@@ -217,11 +227,11 @@ public class ScriptExecutorJobTask implements TimerTask {
JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
//放置调度记录
if (jobTaskRunLogById.getRunCount()>1) {
if (jobTaskRunLogById.getRunCount() > 1) {
//第一次调度的日志
String triggerMsg = jobTaskRunLogById.getTriggerMsg();
jobTaskRunLog.setTriggerMsg(triggerMsg + dispatchResponseDto.getMsg() + "\n");
}else{
} else {
jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg() + "\n");
}
......@@ -233,24 +243,24 @@ public class ScriptExecutorJobTask implements TimerTask {
jobTaskRunLog.setTriggerTime(new Date());
String code = RunResultEnum.TRIGGER_SUCCESS.getCode();
//放置调度码 调度失败 那么执行肯定也失败
if(JobResultEnum.DISPATCH_FAIL.getCode().equals(dispatchResponseDto.getCode())){
if (JobResultEnum.DISPATCH_FAIL.getCode().equals(dispatchResponseDto.getCode())) {
code = RunResultEnum.TRIGGER_ERROR.getCode();
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
jobTaskRunLog.setRunCode(RunResultEnum.RUN_ERROR.getCode());
jobTaskRunLog.setAlertEnd(EmailEnum.IS_ALARM_NO.getCode());
if (jobTaskRunLogById.getRunCount()>1) {
if (jobTaskRunLogById.getRunCount() > 1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg + dispatchResponseDto.getMsg()+LINE);
}else{
jobTaskRunLog.setRunMsg(dispatchResponseDto.getMsg()+LINE);
jobTaskRunLog.setRunMsg(runMsg + dispatchResponseDto.getMsg() + LINE);
} else {
jobTaskRunLog.setRunMsg(dispatchResponseDto.getMsg() + LINE);
}
}
//失败
jobTaskRunLog.setTriggerCode(code);
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount()+1);
jobTaskRunLog.setRunCount(jobTaskRunLogById.getRunCount() + 1);
//获取执行机地址
jobTaskRunLog.setJobGroupIp(dispatchResponseDto.getUrl());
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
......@@ -259,10 +269,11 @@ public class ScriptExecutorJobTask implements TimerTask {
/**
* 保存失败的节点执行信息
*
* @param mythJobTaskSchedule 节点信息
*/
private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule,String code){
private void saveErrorLog(JobTaskSchedule mythJobTaskSchedule, String code) {
FastRunLogService bean = SpringUtil.getBean(FastRunLogService.class);
bean.fastErrorLog(mythJobTaskSchedule,code);
bean.fastErrorLog(mythJobTaskSchedule, code);
}
}
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