Skip to content

B
byit-myth-job
Commit 43de90ab by huangfusuper
Options

集成脚本执行器 对接kill接口

parent 6f0b1eff
Showing with 263 additions and 123 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/model/JobTaskRunLog.java
......@@ -180,7 +180,14 @@ public class JobTaskRunLog implements Serializable {
@ApiModelProperty("执行次数")
private Integer runCount;
/**
* 日志文件的远程路径
*/
@ApiModelProperty("日志的远程路径")
private String logRemotelyPath;
/**
*
*/
private static final long serialVersionUID = 1L;
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java
......@@ -11,10 +11,12 @@ import com.byit.job.exceptions.plugin.PluginException;
import com.byit.job.utils.IpUtil;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.rpc.remoting.invoker.route.LoadBalance;
import com.byit.rpc.remoting.invoker.route.RpcLoadBalance;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
......@@ -37,6 +39,13 @@ public class JavaBeanJobTask implements TimerTask {
@Override
public void run(Timeout timeout) {
log.debug("---------开始交验工作流时否正在运行中------------");
if(checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(),mythJobTaskSchedule.getRunId())){
log.warn("--------------该工作流已经被杀死，不执行-------------------");
return;
}
log.debug("-----------------工作流校验完成-------------");
//获取任务级别 1最低 2最高
String priority = mythJobTaskSchedule.getPriority();
......@@ -133,4 +142,19 @@ public class JavaBeanJobTask implements TimerTask {
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
}
/**
* 校验工作流是否被杀死
* @param flowId
* @param runId
* @return
*/
private boolean checkFlowStatusIsKill(Integer flowId,String runId){
RunRecordingServiceImpl bean = SpringUtil.getBean(RunRecordingServiceImpl.class);
RunRecording runRecordingByFlowIdAndRunId = bean.findRunRecordingByFlowIdAndRunId(flowId, runId);
if(runRecordingByFlowIdAndRunId != null){
return "5".equals(runRecordingByFlowIdAndRunId.getFlowRunResult());
}
return false;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java
......@@ -4,16 +4,22 @@ import com.byit.conf.MythJobAutoConfigure;
import com.byit.enums.NodePropertyEnum;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
import com.byit.model.JobTaskRunLogWithBLOBs;
import com.byit.model.JobTaskSchedule;
import com.byit.model.RunRecording;
import com.byit.service.RunScriptService;
import com.byit.util.FlowContent;
import com.byit.service.impl.JobTaskRunLogServiceImpl;
import com.byit.service.impl.RunRecordingServiceImpl;
import com.byit.util.SpringUtil;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;
import java.util.Date;
/**
* @author Administrator
* @author huangfu
*/
@Slf4j
public class ScriptExecutorJobTask implements TimerTask {
......@@ -26,6 +32,13 @@ public class ScriptExecutorJobTask implements TimerTask {
@Override
public void run(Timeout timeout) throws Exception {
log.debug("---------开始交验工作流时否正在运行中------------");
if(checkFlowStatusIsKill(mythJobTaskSchedule.getFlowId(),mythJobTaskSchedule.getRunId())){
log.warn("--------------该工作流已经被杀死，不执行-------------------");
return;
}
log.debug("-----------------工作流校验完成-------------");
//获取任务级别 1最低 2最高
String priority = mythJobTaskSchedule.getPriority();
......@@ -43,6 +56,25 @@ public class ScriptExecutorJobTask implements TimerTask {
}
/**
* 校验工作流是否被杀死
* @param flowId
* @param runId
* @return
*/
private boolean checkFlowStatusIsKill(Integer flowId,String runId){
RunRecordingServiceImpl bean = SpringUtil.getBean(RunRecordingServiceImpl.class);
RunRecording runRecordingByFlowIdAndRunId = bean.findRunRecordingByFlowIdAndRunId(flowId, runId);
if(runRecordingByFlowIdAndRunId != null){
return "5".equals(runRecordingByFlowIdAndRunId.getFlowRunResult());
}
return false;
}
/**
* 运行任务
* @param mythJobTaskSchedule
*/
private void runJob(JobTaskSchedule mythJobTaskSchedule) {
RunScriptService runScriptService = SpringUtil.getBean(RunScriptService.class);
ScriptDto scriptDto = new ScriptDto();
......@@ -53,6 +85,48 @@ public class ScriptExecutorJobTask implements TimerTask {
scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
scriptDto.setCallbackUrl("http://127.0.0.1:8998/job/callbackRes");
DispatchResponseDto dispatchResponseDto = runScriptService.runScript(scriptDto);
System.out.println(dispatchResponseDto);
saveLog(mythJobTaskSchedule,dispatchResponseDto);
}
private void saveLog(JobTaskSchedule mythJobTaskSchedule, DispatchResponseDto dispatchResponseDto){
log.info("-----------saveLog--保存脚本调度日志开始------------");
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+"|"+dispatchResponseDto.getMsg());
}else{
jobTaskRunLog.setTriggerMsg(dispatchResponseDto.getMsg());
}
jobTaskRunLog.setLogId(mythJobTaskSchedule.getLogId());
jobTaskRunLog.setVersionName(mythJobTaskSchedule.getVersionName());
jobTaskRunLog.setRunType("2");
jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
jobTaskRunLog.setHandlerName(mythJobTaskSchedule.getHandlerName());
jobTaskRunLog.setTriggerTime(new Date());
jobTaskRunLog.setTriggerCode(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(dispatchResponseDto.getCode())?"1":"2");
//放置执行记录
if(!(JobResultEnum.DISPATCH_SUCCESS.getCode().equals(dispatchResponseDto.getCode()))){
Date thisTime = new Date();
jobTaskRunLog.setStartTime(thisTime);
jobTaskRunLog.setEndTime(thisTime);
jobTaskRunLog.setRunCode("2");
jobTaskRunLog.setAlertEnd("1");
if (jobTaskRunLogById.getRunCount()>1) {
//上一次的执行日志
String runMsg = jobTaskRunLogById.getRunMsg();
jobTaskRunLog.setRunMsg(runMsg+"|"+"调度失败");
}
}
//获取执行机地址
jobTaskRunLog.setJobGroupIp(dispatchResponseDto.getUrl());
jobTaskRunLogService.updateJobTaskRunLogWithBLOBs(jobTaskRunLog);
log.info("-----------saveLog--保存脚本调度日志结束------------");
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/LogCallbackThread.java
......@@ -32,11 +32,15 @@ public class LogCallbackThread implements Runnable {
if (jobTaskRunLogById.getRunCount()>1) {
String runMsg = jobTaskRunLogById.getRunMsg()+"|"+jobRunResultDto.getReturnResult().getMsg();
//这里需要追加文件 TODO
//这里需要追加文件 TODO
jobTaskRunLog.setRunMsg(runMsg);
}else{
jobTaskRunLog.setRunMsg(jobRunResultDto.getReturnResult().getMsg());
}
jobTaskRunLog.setLogRemotelyPath(jobRunResultDto.getLogRemotelyPath());
jobTaskRunLog.setLogId(jobRunResultDto.getLogId());
jobTaskRunLog.setStartTime(jobRunResultDto.getStartTime());
jobTaskRunLog.setEndTime(jobRunResultDto.getEndTime());
......
byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml
......@@ -29,6 +29,7 @@
<result column="log_file_name" jdbcType="VARCHAR" property="logFileName" />
<result column="super_success_run" jdbcType="CHAR" property="superSuccessRun"/>
<result column="run_count" jdbcType="INTEGER" property="runCount"/>
<result column="log_remotely_path" jdbcType="VARCHAR" property="logRemotelyPath"/>
</resultMap>
<resultMap extends="BaseResultMap" id="ResultMapWithBLOBs" type="com.byit.model.JobTaskRunLogWithBLOBs">
<result column="run_msg" jdbcType="LONGVARCHAR" property="runMsg" />
......@@ -38,13 +39,13 @@
log_id, failed_remaining_count, version_name, flow_id, flow_name, job_group_id, handler_name,
node_name, is_virtual, run_code, run_params, start_time, run_type, trigger_code,
trigger_time, job_group_ip, map_flow_id, run_command, end_time, node_id,job_type,alert_end,run_id,re_run_id
,log_file_name, super_success_run, run_count
,log_file_name, super_success_run, run_count, log_remotely_path
</sql>
<sql id="Blob_Column_List">
run_msg, trigger_msg
</sql>
<!--这个差的是啥？ 根据runId 和 上级节点的集合 查询所有的上级节点-->
<!--根据runId 和 上级节点的集合 查询所有的上级节点-->
<select id="findJobTaskRunLogNotEndNodeByRunCodeCount" resultMap="BaseResultMap">
select <include refid="Base_Column_List" />
from job_task_run_log
......@@ -194,6 +195,9 @@
<if test="runCount != null">
run_count,
</if>
<if test="logRemotelyPath != null">
log_remotely_path,
</if>
</trim>
<trim prefix="values (" suffix=")" suffixOverrides=",">
<if test="logId != null">
......@@ -283,6 +287,9 @@
<if test="runCount != null">
#{runCount,jdbcType=INTEGER},
</if>
<if test="logRemotelyPath != null">
#{logRemotelyPath,jdbcType=VARCHAR},
</if>
</trim>
</insert>
<update id="updateJobTaskRunLogWithBLOBs" parameterType="com.byit.model.JobTaskRunLogWithBLOBs">
......@@ -372,6 +379,9 @@
<if test="superSuccessRun != null">
run_count = #{runCount,jdbcType=INTEGER},
</if>
<if test="logRemotelyPath != null">
log_remotely_path = #{logRemotelyPath,jdbcType=VARCHAR},
</if>
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
......@@ -456,6 +466,9 @@
<if test="runCount != null">
run_count = #{runCount,jdbcType=INTEGER},
</if>
<if test="logRemotelyPath != null">
log_remotely_path = #{logRemotelyPath,jdbcType=VARCHAR},
</if>
</set>
where log_id = #{logId,jdbcType=INTEGER}
</update>
......
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/DispatchResponseDto.java
......@@ -6,7 +6,7 @@ import java.io.Serializable;
/**
* @program: byit-myth-job->DispatchResponseDto
* @description: TODO
* @description: 运行结果返回
* @author: huangfu
* @date: 2019/12/23 11:12
**/
......@@ -19,4 +19,5 @@ public class DispatchResponseDto implements Serializable {
private String code;
private String msg;
private String content;
private String url;
}
byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/JobRunResultDto.java
......@@ -35,7 +35,13 @@ public class JobRunResultDto {
* 结束时间
*/
private Date endTime;
/**
* 日志ID
*/
private Integer logId;
/**
* 日志的远程路径
*/
private String logRemotelyPath;
}
byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/conf/FileSystemContext.java deleted 100644 → 0
package com.byit.executor.conf;
import com.byit.executor.filesystem.FileSystem;
import org.apache.commons.io.FileUtils;
import org.csource.common.MyException;
import java.io.IOException;
import java.util.Map;
/**
* 文件系统上下文对象
* @author huangfu
*/
public class FileSystemContext {
private FileSystem fileSystem;
public FileSystemContext(FileSystem fileSystem) {
this.fileSystem = fileSystem;
}
public String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTA) throws IOException, MyException {
return fileSystem.uploadFile(fileBuffer,fileExtName,mateDaTA);
}
public byte[] downloaderFile(String remPath) throws IOException, MyException {
return fileSystem.downloaderFile(remPath);
}
public void fileRemove(String remPath) throws IOException, MyException {
fileSystem.fileRemove(remPath);
}
public Map<String, String> getFileMate(String filePath) throws IOException, MyException {
return fileSystem.getFileMate(filePath);
}
}
byit-myth-core/myth-executor-core/src/test/java/com/test/FastDFSTest.java deleted 100644 → 0
package com.test;
import com.byit.executor.conf.FileSystemContext;
import com.byit.executor.filesystem.FastDfsFileSystem;
import org.apache.commons.io.FileUtils;
import org.csource.common.MyException;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerServer;
import org.junit.Before;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class FastDFSTest {
private FileSystemContext fileSystemContext;
@Before
public void init(){
fileSystemContext = new FileSystemContext(new FastDfsFileSystem());
}
@Test
public void testUpload() throws IOException, MyException {
byte[] bytes = FileUtils.readFileToByteArray(new File("E:\\image/20180101.log"));
Map<String,String> map = new HashMap<String,String>(10);
map.put("filename","20180101.log");
//ddmp/M00/00/00/CgB4Al5GY5GAMGVaAAAHa9wlKKc755.log
System.out.println(fileSystemContext.uploadFile(bytes, "log", map));
}
@Test
public void downLoad() throws IOException, MyException {
String filePath = "ddmp/M00/00/00/CgB4Al5J-4aAQ8v6AAAHa95jlv4140.log";
Map<String, String> fileMate = fileSystemContext.getFileMate(filePath);
String filename = fileMate.get("filename");
byte[] bytes = fileSystemContext.downloaderFile(filePath);
File file = new File("E:\\image\\down/" + filename);
if(file.exists()){
file.delete();
}
FileUtils.writeByteArrayToFile(file,bytes);
}
@Test
public void removeTest() throws IOException, MyException {
fileSystemContext.fileRemove("ddmp/M00/00/00/CgB4Al5J-4aAQ8v6AAAHa95jlv4140.log");
}
}
byit-myth-core/myth-executor-core/src/test/java/com/test/Test.java
......@@ -19,14 +19,12 @@ public class Test {
List<MythJobProcess> list = new ArrayList<>();
Thread thread = new Thread(() -> {
List<String> cmdList = new ArrayList<>();
cmdList = Arrays.asList("python D:\\workSpace\\python\\test\\test1.py".split(" "));
Map<String, String> env = new HashMap<>();
env.put("python", "C:\\Program Files\\Python38");
List<String> cmdList = Arrays.asList("python D:\\2020project\\byit-myth-job\\byit-myth-core\\myth-executor-core\\src\\test\\java\\com\\test\\test-ex.py".split(" "));;
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, 0);
list.add(mythJobProcess);
String log = mythJobProcess.call();
System.out.println("运行日志" + log);
System.out.println("运行日志" + log+"--");
System.out.println(mythJobProcess.getProcessId());
});
thread.start();
......@@ -40,4 +38,7 @@ public class Test {
}
}
byit-myth-core/myth-executor-core/src/test/java/com/test/Test1.java 0 → 100644
package com.test;
import com.byit.executor.filesystem.FastDfsFileSystem;
import org.apache.commons.io.FileUtils;
import org.csource.common.MyException;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class Test1 {
public static void main(String[] args) throws IOException, MyException {
FastDfsFileSystem fds = new FastDfsFileSystem();
byte[] bytes = FileUtils.readFileToByteArray(new File("D:\\2020project\\byit-myth-job\\byit-myth-core\\myth-executor-core\\src\\test\\java\\com\\test\\test-ex.py"));
Map<String,String> map = new HashMap<>();
map.put("filename","test-ex.py");
System.out.println(fds.uploadFile(bytes, "py", map));
}
}
byit-myth-core/myth-executor-core/src/test/java/com/test/test-ex.py 0 → 100644
i = 0
while i<100:
print("--------%d" %i)
i+=1
\ No newline at end of file
byit-myth-executor/myth-executor-api/pom.xml
......@@ -13,4 +13,12 @@
<artifactId>myth-executor-api</artifactId>
<dependencies>
<!--执行器核心控制-->
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-core</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
byit-myth-executor/myth-executor-api/src/main/java/com/byit/executor/api/ScriptExecutorService.java
......@@ -2,7 +2,6 @@ package com.byit.executor.api;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.ScriptDto;
import com.byit.job.enums.JobResultEnum;
/**
* Python等脚本的执行器服务类
......
byit-myth-executor/myth-executor-server/pom.xml
......@@ -15,6 +15,11 @@
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-api</artifactId>
</dependency>
......@@ -34,6 +39,8 @@
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java
......@@ -3,6 +3,8 @@ package com.byit.service;
import com.alibaba.fastjson.JSON;
import com.byit.executor.api.ScriptExecutorService;
import com.byit.executor.filesystem.FileSystem;
import com.byit.executor.jobExecutor.process.MythJobProcess;
import com.byit.executor.jobExecutor.process.ProcessFailureException;
import com.byit.job.dto.DispatchResponseDto;
import com.byit.job.dto.JobRunResultDto;
import com.byit.job.dto.ScriptDto;
......@@ -10,6 +12,7 @@ import com.byit.job.enums.JobResultEnum;
import com.byit.job.vo.ReturnResult;
import com.byit.pool.RunThreadPool;
import com.byit.rpc.remoting.provider.annotation.RpcService;
import com.byit.utils.ServiceInfoUtil;
import lombok.extern.slf4j.Slf4j;
import org.csource.common.MyException;
import org.springframework.beans.factory.annotation.Value;
......@@ -19,9 +22,7 @@ import java.io.*;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
import java.util.*;
/**
* 脚本执行的实现
......@@ -36,10 +37,12 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
private String rootLogPath;
@Value("${myth-job.script.root.path}")
private String rootScriptPath;
private final ServiceInfoUtil serviceInfoUtil;
private final FileSystem fileSystem;
public ScriptExecutorServiceImpl(FileSystem fileSystem) {
public ScriptExecutorServiceImpl(ServiceInfoUtil serviceInfoUtil, FileSystem fileSystem) {
this.serviceInfoUtil = serviceInfoUtil;
this.fileSystem = fileSystem;
}
......@@ -60,12 +63,30 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
String remotePath = scriptDto.getRemotePath();
String command = scriptDto.getCommand();
String scriptPath = byteArrayToFile(remotePath);
command = command + " " + scriptPath;
log.info("-----------执行命令，{}-----------",command);
//开始执行脚本
List<String> cmdList = Arrays.asList(command,scriptPath);
MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId());
String logData = mythJobProcess.call();
byte[] logDataByte = stringToByteArray(logData);
Map<String,String> fileMateData = new HashMap<String,String>(2);
fileMateData.put("filename",scriptDto.getRunId()+scriptDto.getRunId()+".log");
String logPath = "";
try {
//上传日志文件
logPath = fileSystem.uploadFile(logDataByte,"log",fileMateData);
jobRunResultDto.setReturnResult(ReturnResult.SUCCESS);
} catch (ProcessFailureException ignored){
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
} catch (IOException | MyException e) {
jobRunResultDto.setReturnResult(ReturnResult.FAIL);
e.printStackTrace();
}
log.info("-----------执行命令，{},{}-----------",command,scriptPath);
//设置结束时间
jobRunResultDto.setEndTime(new Date());
jobRunResultDto.setReturnResult(ReturnResult.SUCCESS);
jobRunResultDto.setLogId(scriptDto.getLogId());
//设置远程日志文件的路径
jobRunResultDto.setLogRemotelyPath(logPath);
cn.hutool.http.HttpUtil.post(callbackUrl, JSON.toJSONString(jobRunResultDto));
log.info("--------------runPythonScript,脚本调用结束-----------");
});
......@@ -80,6 +101,8 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
.msg(JobResultEnum.DISPATCH_FAIL.getMsg())
.build();
}
dispatchResponseDto.setUrl(ServiceInfoUtil.getIpAndPort());
return dispatchResponseDto;
}
......
byit-myth-executor/myth-executor-server/src/main/java/com/byit/utils/ServiceInfoUtil.java 0 → 100644
package com.byit.utils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.embedded.EmbeddedServletContainerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
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-executor/pom.xml
......@@ -19,19 +19,11 @@
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--执行器核心控制-->
<dependency>
<groupId>myth-job</groupId>
<artifactId>myth-executor-core</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/reference/RpcReferenceBean.java
......@@ -185,6 +185,7 @@ public class RpcReferenceBean {
if (addressSet==null || addressSet.size()==0) {
// pass
} else if (addressSet.size()==1) {
//TODO 这里就是根据路由规则选出的路由地址
finalAddress = addressSet.first();
} else {
finalAddress = loadBalance.rpcInvokerRouter.route(serviceKey, addressSet);
......
demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java
......@@ -70,7 +70,7 @@ public class ScriptAddFlow {
pluginNode2.setType("node");
pluginNode2.setAuthor("皇甫");
pluginNode2.setJobType("SCRIPT");
pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5PswCAGEeiAAAHa95jlv4490.log");
pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5T7xuACemOAAAAOeUjOqs3779.py");
pluginNode2.setRunCommand("python");
pluginNode2.setRunParam("test1");
pluginNodeConfig2.setFailedRetryCount(2);
......
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