Skip to content

B
byit-myth-job
Commit ee35f83e by huangfusuper
Options

修改现有BUG

parent c585aa1f
Showing with 209 additions and 59 deletions
byit-myth-admin/src/main/java/com/byit/config/DemoRegisterConfig.java
......@@ -26,8 +26,8 @@ public class DemoRegisterConfig {
private String biz;
@Value("${myth-rpc.registry.env}")
private String env;
@Value("${myth-rpc.registry.port}")
private int port;
// @Value("${myth-rpc.registry.port}")
// private int port;
@Bean
public RpcSpringInvokerFactory invokerFactory(){
......@@ -46,7 +46,7 @@ public class DemoRegisterConfig {
@Bean
public RpcSpringProviderFactory rpcSpringProviderFactory() {
RpcSpringProviderFactory providerFactory = new RpcSpringProviderFactory();
providerFactory.setPort(port);
//providerFactory.setPort(port);
providerFactory.setServiceRegistryClass(RegistryServiceRegistry.class);
providerFactory.setServiceRegistryParam(new HashMap<String, String>() {{
put(RegistryServiceRegistry.REGISTRY_ADDRESS, address);
......
byit-myth-admin/src/main/resources/application-dev.yml
......@@ -22,7 +22,7 @@ myth-rpc:
address: http://localhost:8080/myth-register
env: huangfu
biz: byit-myth-job
port: 6665
#port: 6665
logging:
path: /data/mythjob
file: myth_log_file
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/NodeRunStatusPropertyEnum.java
......@@ -11,7 +11,8 @@ public enum NodeRunStatusPropertyEnum {
RE_RUN_SUCCESS("3","补批成功"),
RE_RUN_FAILURE("4","补批失败"),
KILL("5","kill"),
PARENT_NODE_FAILED("6","上级节点执行失败")
PARENT_NODE_FAILED("6","上级节点执行失败"),
NODE_RELY_ERROR("7","节点依赖错误"),
;
private String code;
private String msg;
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/RunNodeServiceImpl.java
package com.byit.service.impl;
import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodeNameEnum;
import com.byit.event.FlowScanEndEvent;
import com.byit.job.utils.CronExpression;
import com.byit.model.Flow;
import com.byit.model.JobTask;
import com.byit.model.Node;
import com.byit.model.RunRecording;
import com.byit.service.FlowService;
import com.byit.service.JobTaskService;
import com.byit.service.RunNodeServer;
import com.byit.service.RunRecordingService;
import com.byit.service.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
......@@ -36,12 +34,17 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
private final JobTaskService jobTaskService;
private final FlowService flowService;
private ApplicationEventPublisher applicationEventPublisher;
/**
* 节点依赖查询操作
*/
private final NodeDependencyService nodeDependencyService;
public RunNodeServiceImpl(RunRecordingService runRecordingService, JobTaskService jobTaskService,
FlowService flowService) {
FlowService flowService, NodeDependencyService nodeDependencyService) {
this.runRecordingService = runRecordingService;
this.jobTaskService = jobTaskService;
this.flowService = flowService;
this.nodeDependencyService = nodeDependencyService;
}
/**
......@@ -58,7 +61,7 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
BeanUtils.copyProperties(flow,build);
build.setRunId(runId);
//TODO 这个不解释 不知道干嘛的 后续需要修改
build.setDispatchIp("127.0.0.1");
build.setDispatchIp("0.0.0.0");
build.setFlowVersionName(flow.getVersionName());
build.setTriggerTime(flow.getTriggerNextTime());
runRecordingService.saveRunRecording(build);
......@@ -75,6 +78,12 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
}
jobTask.setRunId(runId);
jobTask.setFlowName(flow.getFlowName());
//如果不是开始节点
if(!NodeNameEnum.START_NODE.getNodeName().equals(jobTask.getNodeName())){
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(jobTask.getNodeId());
String parentIds = StringUtils.join(dependIdByNodeId, ",");
jobTask.setNodeDepend(parentIds);
}
jobTasks.add(jobTask);
});
jobTaskService.saveJobTasks(jobTasks);
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -14,6 +14,7 @@ import com.byit.rpc.util.RpcException;
import com.byit.service.RunScriptService;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
import com.byit.util.ServiceInfoUtil;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
......@@ -27,6 +28,9 @@ import java.util.Date;
@Slf4j
public class ScriptExecutorJobTask implements TimerTask {
private final String HTTP_PRE = "http://";
private final String HTTP_SUFFIX = "/job/callbackRes";
private JobTaskSchedule mythJobTaskSchedule;
public ScriptExecutorJobTask(JobTaskSchedule mythJobTaskSchedule) {
......@@ -90,7 +94,7 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto.setParam(mythJobTaskSchedule.getRunParam());
scriptDto.setRunId(mythJobTaskSchedule.getRunId());
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
scriptDto.setCallbackUrl("http://127.0.0.1:8998/job/callbackRes");
scriptDto.setCallbackUrl(HTTP_PRE+ ServiceInfoUtil.getIpAndPort()+HTTP_SUFFIX);
//二次执行的情况下 会有这个信息
scriptDto.setLogRemotePath(jobTaskRunLogById.getLogRemotelyPath());
DispatchResponseDto dispatchResponseDto = new DispatchResponseDto();
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
package com.byit.thread.helper;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.enums.NodeTypeEnum;
import com.byit.job.WorkRoulette;
import com.byit.job.dto.ScriptParamAndPlaceholderDto;
import com.byit.job.enums.PlaceholderEnum;
import com.byit.job.utils.DateUtil;
import com.byit.job.utils.PlaceholderUtils;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.service.JobTaskScheduleService;
......@@ -16,7 +21,9 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.Map;
/**
* 排期表操作
......@@ -48,6 +55,7 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
log.info("------排期表查询到有需要存在的节点--------");
//循环遍历添加任务
jobTaskSchedules.forEach(mythJobTaskSchedule ->{
mythJobTaskSchedule.setRunParam(PlaceholderUtils.formatParam(mythJobTaskSchedule.getRunParam()));
//如果是重跑就有logId
Integer logId = mythJobTaskSchedule.getLogId();
if(logId == null){
......@@ -108,4 +116,5 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
return jobTaskRunLog.getLogId();
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java
......@@ -5,19 +5,20 @@ import com.byit.enums.FlowPropertyEnum;
import com.byit.enums.NodeNameEnum;
import com.byit.enums.NodePropertyEnum;
import com.byit.enums.NodeRunStatusPropertyEnum;
import com.byit.job.exceptions.BusinessException;
import com.byit.model.JobTask;
import com.byit.model.JobTaskRunLog;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.service.JobTaskRunLogService;
import com.byit.service.JobTaskService;
import com.byit.service.NodeDependencyService;
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.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Component;
......@@ -49,10 +50,7 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
* 运行记录表信息操作
*/
private final RunRecordingService runRecordingService;
/**
* 节点依赖查询操作
*/
private final NodeDependencyService nodeDependencyService;
/**
* 日志节点操作
*/
......@@ -68,26 +66,28 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
public TaskThreadRunHelper(DataSource dataSource, JobTaskService jobTaskService,
RunRecordingAndJobTaskService runRecordingAndJobTaskService,
RunRecordingService runRecordingService, NodeDependencyService nodeDependencyService,
JobTaskRunLogService jobTaskRunLogService, TaskAndLogServer taskAndLogServer,
TaskAndScheduleService taskAndScheduleService) {
RunRecordingService runRecordingService, JobTaskRunLogService jobTaskRunLogService,
TaskAndLogServer taskAndLogServer, TaskAndScheduleService taskAndScheduleService) {
this.dataSource = dataSource;
this.jobTaskService = jobTaskService;
this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
this.runRecordingService = runRecordingService;
this.nodeDependencyService = nodeDependencyService;
this.jobTaskRunLogService = jobTaskRunLogService;
this.taskAndLogServer = taskAndLogServer;
this.taskAndScheduleService = taskAndScheduleService;
}
/**
* 业务逻辑
* @return 睡眠时间
*/
@Override
public Long start() {
long nowTime = System.currentTimeMillis();
//开始寻找此时 不是暂停状态，而且七秒内即将运行的任务 而且还不是暂停的节点
List<JobTask> jobTasks = jobTaskService.findJobTaskByTriggerNextTimeLessThanEqual(nowTime + PRE_READ_MS);
if(CollectionUtil.isNotEmpty(jobTasks)) {
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<JobTaskSchedule>(15);
List<JobTaskSchedule> jobTaskSchedules = new ArrayList<>(15);
//遍历七秒内将要运行的节点数据
for(JobTask jobTask : jobTasks){
log.debug("任务：{}，开始运行", jobTask);
......@@ -130,10 +130,21 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
*/
private void nodeOperating(JobTask thisJobTask,List<JobTaskSchedule> jobTaskSchedules){
//查询该节点的依赖节点
List<Integer> dependIdByNodeId = nodeDependencyService.findDependIdByNodeId(thisJobTask.getNodeId());
List<Integer> dependIdByNodeId = null;
if(StringUtils.isNotBlank(thisJobTask.getNodeDepend())){
dependIdByNodeId = new ArrayList<>();
String[] parentIds = thisJobTask.getNodeDepend().split(",");
for (String parentId : parentIds) {
dependIdByNodeId.add(Integer.parseInt(parentId));
}
}
//这里返回的是上级节点的日志执行情况 把运行中的数据给过滤掉了
List<JobTaskRunLog> jobTaskRunLogList = jobTaskRunLogService.findJobTaskRunLogNotEndNodeByRunCodeCount(dependIdByNodeId, thisJobTask.getRunId());
if (CollectionUtil.isNotEmpty(jobTaskRunLogList)) {
if(CollectionUtil.isEmpty(dependIdByNodeId)){
throw new BusinessException(NodeRunStatusPropertyEnum.NODE_RELY_ERROR.getMsg());
}
//判断父类节点是否已经全部完成，只需要判断依赖节点的数目和查询出来的日志数据是否相同
if(dependIdByNodeId.size() == jobTaskRunLogList.size()){
//过滤失败的节点
......@@ -209,7 +220,7 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
/**
* 判断失败节点的重试次数是不是为0
* @param errorJobLog 上级节点的全部失败节点
* @return
* @return 失败节点是否有重试次数
*/
private boolean parentNodeErrorCount(List<JobTaskRunLog> errorJobLog){
if(CollectionUtil.isEmpty(errorJobLog)){
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/ServiceInfoUtil.java 0 → 100644
package com.byit.util;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.embedded.EmbeddedServletContainerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import java.net.InetAddress;
import java.net.UnknownHostException;
/**
* 获取当前服务器的信息数据
* @author huangfu
*/
@Component
@Slf4j
public class ServiceInfoUtil implements ApplicationListener<EmbeddedServletContainerInitializedEvent> {
private static EmbeddedServletContainerInitializedEvent event;
@Override
public void onApplicationEvent(EmbeddedServletContainerInitializedEvent embeddedServletContainerInitializedEvent) {
ServiceInfoUtil.event = embeddedServletContainerInitializedEvent;
}
public static EmbeddedServletContainerInitializedEvent getEvent() {
return event;
}
public static String getIpAndPort() {
int port = getEvent().getEmbeddedServletContainer().getPort();
Assert.state(port != -1, "端口号获取失败");
InetAddress address = null;
try {
address = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
e.printStackTrace();
}
assert address != null;
return address.getHostAddress() +":"+port;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/enums/PlaceholderEnum.java 0 → 100644
package com.byit.job.enums;
/**
* 替换参数枚举
* @author huangfu
*/
public enum PlaceholderEnum {
DATE_PLACEHOLDER("biz_date"),
BIZ_SCRIPT_FILE("biz_file")
;
private String name;
PlaceholderEnum(String name) {
this.name = name;
}
PlaceholderEnum() {
}
public String getName() {
return name;
}
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/PlaceholderUtils.java
......@@ -3,15 +3,10 @@ package com.byit.job.utils;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.byit.job.dto.ScriptParamAndPlaceholderDto;
import lombok.extern.java.Log;
import com.byit.job.enums.PlaceholderEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.HashMap;
......@@ -32,16 +27,12 @@ public class PlaceholderUtils {
* 占位符后缀
*/
private static final String PLACEHOLDER_SUFFIX = "}";
/**
* 补批时间参数占位符
*/
private static final String DATE_KEY = "biz_date";
/**
* 占位符数据替换
* @param scriptData
* @param parameter
* @return
* @param scriptData 脚本数据
* @param parameter 脚本参数
* @return 替换后的脚本参数
*/
public static byte[] resolvePlaceholders(byte[] scriptData, Map<String,String> parameter){
//替换为null 不做操作
......@@ -63,7 +54,7 @@ public class PlaceholderUtils {
//截取变量值
String placeholder = sbt.substring(startIndex + PLACEHOLDER_PREFIX.length(), endIndex);
//确定下一次查询的位置
int nextIndex = endIndex+PLACEHOLDER_SUFFIX.length();
int nextIndex;
//开始从 map中解析
if(parameter.containsKey(placeholder)){
String placeholderValue = parameter.get(placeholder);
......@@ -80,37 +71,84 @@ public class PlaceholderUtils {
return sbt.toString().getBytes(StandardCharsets.UTF_8);
}
/**
* 命令占位符替换
* @param command 命令
* @param param 参数
* @return 替换后的参数
*/
public static String commandReplace(String command,Map<String,String> param){
if(StringUtils.isBlank(command) || CollectionUtil.isEmpty(param)){
log.info("-----------命令{}，不需要替换-----------",command);
return command;
}
byte[] bytes = resolvePlaceholders(command.getBytes(), param);
return new String(bytes, StandardCharsets.UTF_8);
}
/**
* 脚本时间参数替换
* @param runParam 运行参数
* @param thisDate 补批时间
* @return 替换后的参数
*/
public static String paramPlaceholder(String runParam, String thisDate){
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(runParam, ScriptParamAndPlaceholderDto.class);
Map<String, String> placeholder = scriptParamAndPlaceholderDto.getPlaceholder();
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
/**
* 替换时间参数
*/
if (CollectionUtil.isNotEmpty(placeholder) && placeholder.containsKey(DATE_KEY)) {
placeholder.put(DATE_KEY,thisDate);
//替换时间参数
if (CollectionUtil.isNotEmpty(placeholder) && placeholder.containsKey(PlaceholderEnum.DATE_PLACEHOLDER.getName())) {
placeholder.put(PlaceholderEnum.DATE_PLACEHOLDER.getName(),thisDate);
}
/**
* 替换时间占位符
*/
if(CollectionUtil.isNotEmpty(param) && param.containsKey(DATE_KEY)){
param.put(DATE_KEY,thisDate);
//替换时间占位符
if(CollectionUtil.isNotEmpty(param) && param.containsKey(PlaceholderEnum.DATE_PLACEHOLDER.getName())){
param.put(PlaceholderEnum.DATE_PLACEHOLDER.getName(),thisDate);
}
return JSON.toJSONString(scriptParamAndPlaceholderDto);
}
/**
* 初始化脚本信息
* @param command 基础命令
* @param scriptPath 脚本路径
* @return 基础命令
*/
public static String initCommand(String command,String scriptPath){
log.info("----------替换路径{}------",scriptPath);
return command.replace(PLACEHOLDER_PREFIX+PlaceholderEnum.BIZ_SCRIPT_FILE.getName()+PLACEHOLDER_SUFFIX,scriptPath);
}
/**
* 格式化运行参数
* 如果参数内存在时间参数 就将时间参数改为
* @param runParam 运行参数
* @return 参数字符串
*/
public static String formatParam(String runParam){
if (StringUtils.isBlank(runParam)) {
return null;
}
ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = JSON.parseObject(runParam,ScriptParamAndPlaceholderDto.class);
Map<String, String> param = scriptParamAndPlaceholderDto.getParam();
if(CollectionUtil.isNotEmpty(param) && param.containsKey(PlaceholderEnum.DATE_PLACEHOLDER.getName())){
String calculationDate = DateUtil.dateLessDayStr(new Date(), "yyyyMMdd", 1);
param.put(PlaceholderEnum.DATE_PLACEHOLDER.getName(),calculationDate);
}
return JSON.toJSONString(scriptParamAndPlaceholderDto);
}
public static void main(String[] args) {
ScriptParamAndPlaceholderDto sc = new ScriptParamAndPlaceholderDto();
/*ScriptParamAndPlaceholderDto sc = new ScriptParamAndPlaceholderDto();
Map<String,String> m1 = new HashMap<>(2);
Map<String,String> m2 = new HashMap<>(2);
m1.put(DATE_KEY,"2018年12月12日");
m1.put(PlaceholderEnum.DATE_PLACEHOLDER.getName(),"2018年12月12日");
m1.put("name","狗子");
m2.put(DATE_KEY,"2010年10月10日");
m2.put(PlaceholderEnum.DATE_PLACEHOLDER.getName(),"2010年10月10日");
m2.put("age","1000");
......@@ -118,13 +156,13 @@ public class PlaceholderUtils {
sc.setParam(m1);
String toJSONString = JSON.toJSONString(sc);
String jsonString = JSON.toJSONString(sc);
String s = PlaceholderUtils.paramPlaceholder(toJSONString, "2020/3/12 15:13");
String s = PlaceholderUtils.paramPlaceholder(jsonString, "2020/3/12 15:13");*/
System.out.println(initCommand("python ${biz_file}","D:/2020project/byit-myth-job/demo-client/byit-demo-client/src/main/java/com/byit/job/Mains.java"));
System.out.println(s);
}
}
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
......@@ -73,8 +73,14 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
}
String scriptPath = byteArrayToFile(remotePath,scriptParamAndPlaceholderDto);
log.info("-------------脚本所在路径为：{}-------------",scriptPath);
//初始化命令信息
command = PlaceholderUtils.initCommand(command,scriptPath);
//开始执行脚本
List<String> cmdList = Arrays.asList(command,scriptPath);
if(scriptParamAndPlaceholderDto != null){
command = PlaceholderUtils.commandReplace(command,scriptParamAndPlaceholderDto.getParam());
}
List<String> cmdList = Arrays.asList(command.split(" "));
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId());
//保存日志
String logData = mythJobProcess.call();
......@@ -207,4 +213,12 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
return date.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
}
public static void main(String[] args) throws IOException {
File file = new File("D:\\2020project\\byit-myth-job\\demo-client\\byit-demo-client\\src\\main\\java\\com\\byit\\job\\Mains.java");
System.out.println(file.getPath());
System.out.println(file.getAbsolutePath());
System.out.println(file.getCanonicalPath());
}
}
demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java
......@@ -77,7 +77,7 @@ public class ScriptAddFlow {
scriptParamAndPlaceholderDto.setPlaceholder(map);
pluginNode2.setScriptParam(scriptParamAndPlaceholderDto);
pluginNode2.setRunCommand("python");
pluginNode2.setRunCommand("python ${biz_file}");
pluginNodeConfig2.setFailedRetryCount(2);
pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
......@@ -96,7 +96,6 @@ public class ScriptAddFlow {
pluginNode5.setAuthor("皇甫");
pluginNode5.setJobType("JAVA");
pluginNode5.setHandlerName("END");
pluginNode5.setRunParam("END");
pluginNodeConfig5.setFailedRetryCount(2);
pluginNodeConfig5.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
pluginNodeConfig5.setNodeCron("0 0/7 * * * ? *");
......
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