Skip to content

B
byit-myth-job
Commit f5fe3b57 by huangfusuper
Options

1. 修正脚本运行命令替换 

2. 添加java任务返回值携带回写参数
parent 5add5521
Showing with 111 additions and 82 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaNodeExecutorTask.java
......@@ -2,9 +2,11 @@ package com.byit.task;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.RunLog;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.job.utils.MythLogUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.packet.request.PluginRpcRequestPacket;
......@@ -14,9 +16,7 @@ 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.job.utils.MythLogUtils;
import com.byit.util.GetRegConfig;
import com.byit.util.ServiceInfoUtil;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
......@@ -32,10 +32,11 @@ import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
/**
* java节点执行
*
* @author huangfu
*/
@Slf4j
public class JavaNodeExecutorTask implements TimerTask {
public class JavaNodeExecutorTask implements TimerTask {
private static final Integer INIT_SLEEP_TIME = 100;
public static final String HTTP_PRE = "http://";
public static final String HTTP_SUFFIX = "/myth-job-admin/api/callback/callbackRes";
......@@ -45,13 +46,14 @@ public class JavaNodeExecutorTask implements TimerTask {
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){
if (code != null) {
log.warn("--------------该工作流已经被杀死快速失败，执行快速失败！-------------------");
saveErrorLog(mythJobTaskSchedule,code);
saveErrorLog(mythJobTaskSchedule, code);
return;
}
log.debug("-----------------工作流校验完成-------------");
......@@ -59,10 +61,10 @@ public class JavaNodeExecutorTask 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));
}
}
......@@ -70,9 +72,13 @@ public class JavaNodeExecutorTask implements TimerTask {
RunJavaServiceImpl service = SpringUtil.getBean(RunJavaServiceImpl.class);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
JobTaskRunLogWithBLOBs jobTaskRunLogById = null;
try{
log.info("--------------logid:{}----------",mythJobTaskSchedule.getLogId());
try {
String runParam = mythJobTaskSchedule.getRunParam();
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
if (runParamWrapped == null) {
runParamWrapped = new RunParamWrapped();
}
log.info("--------------logid:{}----------", mythJobTaskSchedule.getLogId());
jobTaskRunLogById = jobTaskRunLogService.findJobTaskRunLogById(mythJobTaskSchedule.getLogId());
jobTaskRunLogById = spinLock(jobTaskRunLogById);
//构建JAVA
......@@ -85,48 +91,49 @@ public class JavaNodeExecutorTask implements TimerTask {
CommunicationParam communicationParam = new CommunicationParam();
communicationParam.setFlowId(mythJobTaskSchedule.getFlowId());
communicationParam.setHasMakeUp(mythJobTaskSchedule.getScheduleType()+"");
communicationParam.setHasMakeUp(mythJobTaskSchedule.getScheduleType() + "");
communicationParam.setRunId(mythJobTaskSchedule.getRunId());
communicationParam.setExpand1("REAL:EXEC:"+jobTaskRunLogById.getLogId());
communicationParam.setLogId(jobTaskRunLogById.getLogId()+"");
communicationParam.setExpand1("REAL:EXEC:" + jobTaskRunLogById.getLogId());
communicationParam.setLogId(jobTaskRunLogById.getLogId() + "");
communicationParam.setCallbackUrl(JSON.toJSONString(callUrlList));
communicationParam.setBody(mythJobTaskSchedule.getRunParam());
communicationParam.setBody(runParamWrapped.getPrivateParam());
communicationParam.setPublicParam(runParamWrapped.getPublicParamMap());
request.setParam(communicationParam);
request.setJobName(mythJobTaskSchedule.getHandlerName());
request.setExtension(jobTaskRunLogById.getLogId()+"");
request.setExtension(jobTaskRunLogById.getLogId() + "");
jobTaskRunLogById.setLogId(jobTaskRunLogById.getLogId());
jobTaskRunLogById.setStartTime(new Date());
jobTaskRunLogById.setTriggerTime(new Date());
PluginRpcResponsePacket pluginRpcResponsePacket = service.runJava(request);
if(pluginRpcResponsePacket.isStatus()){
if (pluginRpcResponsePacket.isStatus()) {
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_SUCCESS.getCode());
}else{
} else {
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
}
jobTaskRunLogById.setTriggerMsg(pluginRpcResponsePacket.getMsg());
jobTaskRunLogById.setJobGroupIp(pluginRpcResponsePacket.getRunIp());
jobTaskRunLogById.setRunParams(mythJobTaskSchedule.getRunParam());
jobTaskRunLogById.setRunParams(runParam);
jobTaskRunLogById.setRunType("2");
jobTaskRunLogById.setAlertEnd("1");
}catch (Exception e){
} catch (Exception e) {
if (mythJobTaskSchedule.getScheduleType() == 4) {
StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) SpringUtil.getBean("stringRedisTemplate");
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + MythLogUtils.getMessage(e)).build();
RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" + MythLogUtils.getMessage(e)).build();
//stringRedisTemplate.convertAndSend("REAL:EXEC:" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
//使用redis 想队尾push一个值
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));
assert jobTaskRunLogById != null;
jobTaskRunLogById.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
jobTaskRunLogById.setRunCode(RunResultEnum.RUN_ERROR.getCode());
jobTaskRunLogById.setRunMsg(mythJobTaskSchedule.getNodeName()+":"+MythLogUtils.getMessage(e));
jobTaskRunLogById.setTriggerMsg(mythJobTaskSchedule.getNodeName()+":"+MythLogUtils.getMessage(e));
jobTaskRunLogById.setRunMsg(mythJobTaskSchedule.getNodeName() + ":" + MythLogUtils.getMessage(e));
jobTaskRunLogById.setTriggerMsg(mythJobTaskSchedule.getNodeName() + ":" + MythLogUtils.getMessage(e));
}
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLogById);
......@@ -134,24 +141,25 @@ public class JavaNodeExecutorTask implements TimerTask {
/**
* 模拟自旋锁实现
*
* @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;
}
......@@ -161,21 +169,23 @@ public class JavaNodeExecutorTask 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);
}
/**
* 校验工作流是否被杀死或者被快速失败
*
* @param flowId 工作流ID
* @param runId 运行标识
* @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);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaTaskJobTask.java
......@@ -2,19 +2,19 @@ package com.byit.task;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.plugin.JavaTask;
import com.byit.enums.ScheduleTypeEnum;
import com.byit.enums.task.RunResultEnum;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.MythLogUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.packet.request.PluginRpcRequestPacket;
import com.byit.packet.response.PluginRpcResponsePacket;
import com.byit.param.CommunicationParam;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunJavaServiceImpl;
import com.byit.job.utils.MythLogUtils;
import com.byit.util.GetRegConfig;
import com.byit.util.ServiceInfoUtil;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
......@@ -26,6 +26,7 @@ import java.util.stream.Collectors;
/**
* 立即运行 java单任务 java节点的执行器
*
* @author huangfu
*/
public class JavaTaskJobTask implements TimerTask {
......@@ -44,9 +45,14 @@ public class JavaTaskJobTask implements TimerTask {
}
private void runJob(){
private void runJob() {
RunJavaServiceImpl service = SpringUtil.getBean(RunJavaServiceImpl.class);
PluginRpcRequestPacket request = new PluginRpcRequestPacket();
String param = javaTask.getParam();
RunParamWrapped runParamWrapped = JSON.parseObject(param, RunParamWrapped.class);
if (runParamWrapped == null) {
runParamWrapped = new RunParamWrapped();
}
//保存到日志
Integer logId = saveLog(javaTask);
GetRegConfig getRegConfig = SpringUtil.getBean(GetRegConfig.class);
......@@ -55,41 +61,43 @@ public class JavaTaskJobTask implements TimerTask {
request.setCallbackUrl(JSON.toJSONString(callUrlList));
CommunicationParam communicationParam = new CommunicationParam();
communicationParam.setHasMakeUp(javaTask.getScheduleType()+"");
communicationParam.setHasMakeUp(javaTask.getScheduleType() + "");
communicationParam.setLogId(logId+"");
communicationParam.setLogId(logId + "");
communicationParam.setCallbackUrl(JSON.toJSONString(callUrlList));
communicationParam.setBody(javaTask.getParam());
communicationParam.setBody(runParamWrapped.getPrivateParam());
communicationParam.setPublicParam(runParamWrapped.getPublicParamMap());
String dateFormat = DateUtil.dateFormat(new Date(javaTask.getTriggerTime()), DateUtil.FORMAT_DATE_TIME);
communicationParam.setExpand2(dateFormat);
request.setParam(communicationParam);
request.setExtension(logId+"");
request.setExtension(logId + "");
request.setJobName(javaTask.getTaskName());
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
log.setLogId(logId);
try {
PluginRpcResponsePacket pluginRpcResponsePacket = service.runJava(request);
if(pluginRpcResponsePacket.isStatus()){
if (pluginRpcResponsePacket.isStatus()) {
log.setTriggerCode(RunResultEnum.TRIGGER_SUCCESS.getCode());
}else{
} else {
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
}
log.setTriggerMsg(pluginRpcResponsePacket.getMsg());
log.setJobGroupIp(pluginRpcResponsePacket.getRunIp());
}catch (Exception e){
} catch (Exception e) {
log.setTriggerCode(RunResultEnum.TRIGGER_ERROR.getCode());
log.setRunCode(RunResultEnum.RUN_ERROR.getCode());
log.setRunMsg(javaTask.getTaskName()+":"+ MythLogUtils.getMessage(e));
log.setTriggerMsg(javaTask.getTaskName()+":"+MythLogUtils.getMessage(e));
log.setRunMsg(javaTask.getTaskName() + ":" + MythLogUtils.getMessage(e));
log.setTriggerMsg(javaTask.getTaskName() + ":" + MythLogUtils.getMessage(e));
}
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(log);
}
private Integer saveLog(JavaTask javaTask){
private Integer saveLog(JavaTask javaTask) {
JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
log.setStartTime(new Date());
log.setScheduleType(ScheduleTypeEnum.JAVA_SYNC.getCode());
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
package com.byit.task;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.conf.MythJobAutoConfigure;
import com.byit.dto.executor.DispatchResponseDto;
import com.byit.dto.executor.RunParamWrapped;
import com.byit.dto.executor.ScriptDto;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.dto.plugin.RunLog;
import com.byit.enums.*;
import com.byit.enums.task.RunResultEnum;
import com.byit.enums.task.RunTypeEnum;
import com.byit.factory.PluginClientFactory;
import com.byit.factory.PluginSpringClientFactory;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.registry.PluginServiceRegistry;
......@@ -26,10 +30,7 @@ import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.TreeSet;
import java.util.*;
import java.util.stream.Collectors;
import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
......@@ -86,6 +87,40 @@ public class ScriptExecutorJobTask implements TimerTask {
return bean.checkFlowStatusIsKill(flowId,runId);
}
private String commandBuild(JobTaskSchedule mythJobTaskSchedule){
//获取运行时参数
String runParam = mythJobTaskSchedule.getRunParam();
//获取命令
String command = mythJobTaskSchedule.getRunCommand();
//转换参数对象为参数包装体
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
//当参数包装体不为空时 证明存在参数 或私有或公有
if(runParamWrapped != null) {
//获取到私有参数
String privateParam = runParamWrapped.getPrivateParam();
//获取到公有参数
Map<String,String> publicParam = runParamWrapped.getPublicParamMap();
//将私有参数转换为对应的参数DTO
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam,ScriptParamAndPlaceholderDto.class);
//脚本参数不为空的时候
if(scriptParamAndPlaceholderDto != null) {
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
if(CollectionUtil.isNotEmpty(param)){
command = PlaceholderUtils.commandReplace(command,param );
}
}
//公共参数不为空的时候
if(CollectionUtil.isNotEmpty(publicParam)) {
command = PlaceholderUtils.commandReplace(command,publicParam );
}
//替换运行参数中的时间参数
if (ScheduleTypeEnum.NORMAL.getCode().equals(mythJobTaskSchedule.getScheduleType())) {
mythJobTaskSchedule.setRunParam(PlaceholderUtils.formatParam(privateParam));
}
}
return command;
}
/**
* 运行任务
* @param mythJobTaskSchedule 运行的排期表
......@@ -93,6 +128,7 @@ public class ScriptExecutorJobTask implements TimerTask {
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
try{
mythJobTaskSchedule.setRunCommand(commandBuild(mythJobTaskSchedule));
RunScriptService runScriptService = SpringUtil.getBean(RunScriptService.class);
JobTaskRunLogServiceImpl jobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
log.info("--------------logid:{}----------",mythJobTaskSchedule.getLogId());
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
......@@ -62,37 +62,8 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
log.debug("------排期表查询到有需要存在的节点--------");
//循环遍历添加任务
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
//获取运行时参数
String runParam = mythJobTaskSchedule.getRunParam();
//获取命令
String command = mythJobTaskSchedule.getRunCommand();
//转换参数对象为参数包装体
RunParamWrapped runParamWrapped = JSON.parseObject(runParam, RunParamWrapped.class);
//当参数包装体不为空时 证明存在参数 或私有或公有
if(runParamWrapped != null) {
//获取到私有参数
String privateParam = runParamWrapped.getPrivateParam();
//获取到公有参数
Map<String,String> publicParam = runParamWrapped.getPublicParamMap();
//将私有参数转换为对应的参数DTO
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(privateParam,ScriptParamAndPlaceholderDto.class);
//脚本参数不为空的时候
if(scriptParamAndPlaceholderDto != null) {
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
if(CollectionUtil.isNotEmpty(param)){
command = PlaceholderUtils.commandReplace(command,param );
}
}
//公共参数不为空的时候
if(CollectionUtil.isNotEmpty(publicParam)) {
command = PlaceholderUtils.commandReplace(command,publicParam );
}
//替换运行参数中的时间参数
if (ScheduleTypeEnum.NORMAL.getCode().equals(mythJobTaskSchedule.getScheduleType())) {
mythJobTaskSchedule.setRunParam(PlaceholderUtils.formatParam(privateParam));
}
}
//除去JAVA任务
if(!NodeTypeEnum.JAVA.getCode().equals(mythJobTaskSchedule.getJobType())){
assert command != null;
......
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/param/CommunicationParam.java
......@@ -3,6 +3,7 @@ package com.byit.param;
import lombok.Data;
import java.io.Serializable;
import java.util.Map;
/**
* 执行参数
......@@ -39,7 +40,10 @@ public class CommunicationParam implements Serializable {
*/
private String hasMakeUp;
/**
* 设置公共参数
*/
private Map<String,String> publicParam;
/**
* 工作流ID
*/
......
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