Commit 30ccfc81 by huangfusuper

重构部分执行机,优化执行逻辑

parent daf754dc
......@@ -79,7 +79,7 @@ public class PlaceholderUtils {
*/
public static String commandReplace(String command,Map<String,String> param){
if(StringUtils.isBlank(command) || CollectionUtil.isEmpty(param)){
log.info("-----------命令{},不需要替换-----------",command);
log.debug("-----------命令{},不需要替换-----------",command);
return command;
}
StringBuilder sb = new StringBuilder(command);
......
package com.byit.enums;
/**
* 执行机异常信息枚举
* @author huangfu
*/
public enum ExecutorExceptionEnum {
REQUIRED_PARAMETERS_ARE_EMPTY("必要参数为空,请联系开发调度开发团队!"),
EXECUTION_COMMAND_IS_EMPTY("脚本执行命令为空,请联系开发调度开发团队!"),
THE_LOG_INDEX_IS_EMPTY("日志节点为空,请联系开发调度开发团队!"),
THE_CALLBACK_PATH_IS_EMPTY("回调路径为空,请联系开发调度开发团队!"),
THE_REMOTE_SCRIPT_IS_EMPTY("远程脚本为空,请上传远程脚本!")
;
private String msg;
ExecutorExceptionEnum(String msg) {
this.msg = msg;
}
public String getMsg() {
return msg;
}
}
package com.byit.exceptions;
import com.byit.enums.ExecutorExceptionEnum;
/**
* 执行机异常通类
* @author huangfu
*/
public class ExecutorException extends RuntimeException {
public ExecutorException(String msg) {
super(msg);
}
public ExecutorException(Throwable throwable) {
super(throwable);
}
public ExecutorException (ExecutorExceptionEnum executorExceptionEnum) {
super(executorExceptionEnum.getMsg());
}
}
package com.byit.node;
import com.byit.dto.executor.ScriptDto;
import com.byit.node.machine.ProcessingMachine;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 这个节点链条的数据承载
* @author Administrator
*/
@Slf4j
public class ProcessingNodeChain {
/**
* 责任链条
* 用于存储所有的处理节点,使其成为一个链条
*/
private Vector<ProcessingMachine> processingMachines = new Vector<>(8);
/**
* 计数器 用于记录当前的处理节点所在的索引位置
*/
private AtomicInteger index = new AtomicInteger(0);
/**
* 用于向处理链条添加 一个链条:使两条链条连接
* @param processingMachines 一条处理链节点组合
* @return 当前完整的链条规则
*/
public ProcessingNodeChain addProcessingNodes(List<ProcessingMachine> processingMachines){
processingMachines.addAll(processingMachines);
return this;
}
/**
* 此方法用于向链条添加一个连接点 也就是处理节点
* @param processingMachine 要添加的处理节点
* @return 返回完整的链条规则
*/
public ProcessingNodeChain addProcessingNode(ProcessingMachine processingMachine){
processingMachines.add(processingMachine);
return this;
}
/**
* 开始处理
* @param scriptDto 要处理的元数据
* @param processingNodeChain 当前的链条数据
*/
public void doProcessing(ScriptDto scriptDto, ProcessingNodeChain processingNodeChain){
int chainIndex = index.get();
log.debug("------处理到第:{}个位置的节点:{}--------",chainIndex,scriptDto);
if (chainIndex == processingMachines.size()){
return;
}
//获取对应索引的节点数据
ProcessingMachine processingMachine = processingMachines.get(chainIndex);
int nextIndex = index.incrementAndGet();
//真正的执行数据
processingMachine.executor(scriptDto,processingNodeChain);
log.debug("------本节点处理完毕,下一个要处理的节点索引为{}--------",nextIndex);
}
}
package com.byit.node.impl;
import com.alibaba.fastjson.JSON;
import com.byit.dto.executor.ScriptDto;
import com.byit.dto.executor.ScriptParamAndPlaceholderDto;
import com.byit.enums.PlaceholderEnum;
import com.byit.filesystem.FileSystem;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.node.ProcessingNodeChain;
import com.byit.node.machine.ProcessingMachine;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
/**
* 2.命令处理节点,主要做命令的参数替换操作
* @author huangfu
**/
@Component
@Slf4j
public class CommandAndScriptProcessingMachine implements ProcessingMachine {
private final FileSystem fileSystem;
/**
* 脚本生成路径
*/
@Value("${myth-job.script.root.path}")
private String rootScriptPath;
public CommandAndScriptProcessingMachine(FileSystem fileSystem) {
this.fileSystem = fileSystem;
}
@Override
public void executor(ScriptDto scriptDto, ProcessingNodeChain processingNodeChain) {
log.debug("-----执行到命令和脚本的处理节点,开始初始化命令和脚本------------");
String command = scriptDto.getCommand();
log.debug("-----初始命令为{}------------",command);
String param = scriptDto.getParam();
if (StringUtils.isNotBlank(param)) {
ScriptParamAndPlaceholderDto paramAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
//判断是否需要拉取文件服务器的文件,包含biz_file的条件下会对这个字段进行替换
if (command.contains(PlaceholderEnum.BIZ_SCRIPT_FILE.getName())) {
String scriptPath = byteArrayToFile(scriptDto.getRemotePath(),paramAndPlaceholderDto);
log.debug("-----远程地址的文件被拉取到:[{}]------------",scriptPath);
//初始化命令信息
command = PlaceholderUtils.initCommand(command,scriptPath);
log.debug("-----执行命令初始化完成,命令为:[{}]------------",command);
scriptDto.setCommand(command);
}
//执行命令参数的替换
command = PlaceholderUtils.commandReplace(command,paramAndPlaceholderDto.getParam());
log.debug("-----命令参数替换完成,命令为:[{}]------------",command);
scriptDto.setCommand(command);
}
//将脚本的携带的参数 拼接在命令后方 python test.py -sex 男 -name 张三 -age 14
log.info("---------命令及脚本参数处理完成,此节点处理数据为:[{}]-----------",scriptDto);
processingNodeChain.doProcessing(scriptDto,processingNodeChain);
}
/**
* 将脚本字节转换成文件
* @return 生成文件的本地路径
*/
private String byteArrayToFile(String remotePath,ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto){
//创建目录
File rootPathMkdir = new File(rootScriptPath, DateUtil.dateFormat(new Date(),"yyyyMMdd"));
if(!rootPathMkdir.exists()){
rootPathMkdir.mkdirs();
}
OutputStream out = null;
File file;
try {
//下载脚本文件
byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
//替换脚本占位符
if(scriptParamAndPlaceholderDto != null){
scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray,scriptParamAndPlaceholderDto.getPlaceholder());
}
//获取文件元信息
Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
String fileName = fileMate.get("filename");
fileName = UUID.randomUUID().toString().replace("-","")+fileName;
file = new File(rootPathMkdir,fileName);
out = new FileOutputStream(file);
//写出脚本文件
out.write(scriptByteArray);
return file.getPath();
} catch (IOException e) {
e.printStackTrace();
} finally {
if(out != null){
try {
out.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return null;
}
}
package com.byit.node.impl;
import com.byit.dto.executor.ScriptDto;
import com.byit.enums.ExecutorExceptionEnum;
import com.byit.enums.PlaceholderEnum;
import com.byit.exceptions.ExecutorException;
import com.byit.node.ProcessingNodeChain;
import com.byit.node.machine.ProcessingMachine;
import org.apache.commons.lang3.StringUtils;
/**
* 1.数据校验处理节点
* 主要就是校验 scriptDto对象内部参数的合法性,若不合法则直接异常
* @author huangfu
*/
public class ParameterVerificationProcessingMachine implements ProcessingMachine {
@Override
public void executor(ScriptDto scriptDto, ProcessingNodeChain processingNodeChain) {
//参数为null
if (null == scriptDto) {
throw new ExecutorException(ExecutorExceptionEnum.REQUIRED_PARAMETERS_ARE_EMPTY);
}
//执行命令为null
if(StringUtils.isBlank(scriptDto.getCommand())){
throw new ExecutorException(ExecutorExceptionEnum.EXECUTION_COMMAND_IS_EMPTY);
}
//日志id为null
if (null == scriptDto.getLogId()){
throw new ExecutorException(ExecutorExceptionEnum.THE_LOG_INDEX_IS_EMPTY);
}
//运行标识为null
if (StringUtils.isBlank(scriptDto.getRunId())) {
throw new ExecutorException(ExecutorExceptionEnum.THE_LOG_INDEX_IS_EMPTY);
}
//回调路径为null
if (StringUtils.isBlank(scriptDto.getCallbackUrl())) {
throw new ExecutorException(ExecutorExceptionEnum.THE_CALLBACK_PATH_IS_EMPTY);
}
//查看是否存在远程脚本
if (scriptDto.getCommand().contains(PlaceholderEnum.BIZ_SCRIPT_FILE.getName())) {
if (StringUtils.isBlank(scriptDto.getRemotePath())) {
throw new ExecutorException(ExecutorExceptionEnum.THE_REMOTE_SCRIPT_IS_EMPTY);
}
}
// 回调处理链上的下一个节点
processingNodeChain.doProcessing(scriptDto,processingNodeChain);
}
}
package com.byit.node.machine;
import com.byit.dto.executor.ScriptDto;
import com.byit.node.ProcessingNodeChain;
/**
* 处理机器
* @author huangfu
*/
public interface ProcessingMachine {
/**
* 执行数据处理 每一个节点的处理类都会处理这个参数,同时将参数向下传递
* @param scriptDto 核心参数,内部的数据被功能节点处理后,再次向下传递
* @param processingNodeChain 链条
*/
void executor(ScriptDto scriptDto, ProcessingNodeChain processingNodeChain);
}
......@@ -125,6 +125,7 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
if(scriptParamAndPlaceholderDto != null){
command = PlaceholderUtils.commandReplace(command,scriptParamAndPlaceholderDto.getParam());
}
//TODO 链条到这个地步了 考虑异步操作 等待完成 超时报错
List<String> cmdList = Arrays.asList(command.split(" "));
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId(), stringRedisTemplate);
//保存日志
......
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