修改运行的key和报错问题

byit-myth-admin/pom.xml

@@ -53,11 +53,6 @@
             <version>1.0-SNAPSHOT</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-redis</artifactId>
-            <version>1.3.2.RELEASE</version>
-        </dependency>
     </dependencies>
 
     <build>

byit-myth-admin/src/main/java/com/byit/service/impl/ApiNodeServiceImpl.java

@@ -4,10 +4,17 @@ import com.alibaba.fastjson.JSON;
 import com.byit.dto.plugin.RunLog;
 import com.byit.dto.plugin.RunNode;
 import com.byit.enums.NodePropertyEnum;
+import com.byit.job.WorkRoulette;
+import com.byit.job.utils.PlaceholderUtils;
+import com.byit.mapper.JobTaskRunLogMapper;
 import com.byit.mapper.JobTaskScheduleMapper;
+import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.model.JobTaskSchedule;
 import com.byit.service.ApiNodeService;
+import com.byit.task.ScriptExecutorJobTask;
 import com.byit.utils.ValidationUtil;
+import io.netty.util.TimerTask;
+import org.springframework.beans.BeanUtils;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
@@ -22,7 +29,7 @@ import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
 public class ApiNodeServiceImpl implements ApiNodeService {
 
     @Resource
-    private JobTaskScheduleMapper jobTaskScheduleMapper;
+    private JobTaskRunLogMapper jobTaskRunLogMapper;
 
     @Resource
     private StringRedisTemplate stringRedisTemplate;
@@ -39,8 +46,10 @@ public class ApiNodeServiceImpl implements ApiNodeService {
         ValidationUtil.dataNotBank(runNode.getRunCmd(), "运行命令不允许为空！");
         ValidationUtil.dataNotBank(runNode.getJobType(), "节点类型不允许为空！");
 
-        JobTaskSchedule schedule = new JobTaskSchedule();
+        Long triggerTime = System.currentTimeMillis();
+
 
+        JobTaskSchedule schedule = new JobTaskSchedule();
         schedule.setNodeId(Integer.valueOf(runNode.getNodeId()));
         schedule.setRunId(runId);
         schedule.setNodeName(runNode.getNodeName());
@@ -50,13 +59,23 @@ public class ApiNodeServiceImpl implements ApiNodeService {
         schedule.setRunSource(runNode.getRunSource());
         schedule.setScriptUrls(runNode.getScriptUrl());
         schedule.setJobType(runNode.getJobType());
-        schedule.setTriggerTime(System.currentTimeMillis());
+        schedule.setTriggerTime(triggerTime);
 
-        jobTaskScheduleMapper.saveJobTaskSchedule(schedule);
+        schedule.setRunParam(PlaceholderUtils.formatParam(schedule.getRunParam()));
 
+        JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
+        BeanUtils.copyProperties(schedule, jobTaskRunLog);
+
+        jobTaskRunLogMapper.saveJobTaskRunLog(jobTaskRunLog);
+        //获取日志id
+        Integer logId= jobTaskRunLog.getLogId();
+        TimerTask timerTask = new ScriptExecutorJobTask(schedule);
+        WorkRoulette.addJob(timerTask, triggerTime);
         RunLog runLog = RunLog.builder().isEnd(false).runLog("等待服务器分配资源").build();
-        stringRedisTemplate.convertAndSend(runId, JSON.toJSONString(runLog, WriteClassName));
+        String runKey = "REAL-EXEC-" + logId;
+
+        stringRedisTemplate.convertAndSend(runKey, JSON.toJSONString(runLog, WriteClassName));
 
-        return runId;
+        return runKey;
     }
 }

byit-myth-core/myth-admin-core/pom.xml

@@ -70,6 +70,12 @@
             <scope>compile</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-redis</artifactId>
+            <version>1.3.2.RELEASE</version>
+        </dependency>
+
     </dependencies>
 
     <build>

byit-myth-admin/src/main/java/com/byit/config/RedisConfig.java → byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/RedisConfig.java

-package com.byit.config;
+package com.byit.conf;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java

 package com.byit.task;
 
+import com.alibaba.fastjson.JSON;
 import com.byit.conf.MythJobAutoConfigure;
+import com.byit.dto.plugin.RunLog;
 import com.byit.enums.EmailEnum;
 import com.byit.enums.NodePropertyEnum;
 import com.byit.enums.NodeRunStatusPropertyEnum;
@@ -19,9 +21,12 @@ import com.byit.util.SpringUtil;
 import io.netty.util.Timeout;
 import io.netty.util.TimerTask;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.StringRedisTemplate;
 
 import java.util.Date;
 
+import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
+
 /**
  * @author huangfu
  */
@@ -101,7 +106,10 @@ public class ScriptExecutorJobTask implements TimerTask {
         try{
             dispatchResponseDto = runScriptService.runScript(scriptDto);
         }catch (RpcException rpcException){
-            log.error("------执行机异常{}-----",rpcException.getMessage());
+            StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) SpringUtil.getBean("stringRedisTemplate");
+            RunLog runLog = RunLog.builder().isEnd(true).isSuccess(false).runLog("执行资源异常" +  rpcException.getMessage()).build();
+            stringRedisTemplate.convertAndSend("REAL-EXEC-" + mythJobTaskSchedule.getLogId() , JSON.toJSONString(runLog, WriteClassName));
+            log.error("------执行机异常{}-----", rpcException.getMessage());
             dispatchResponseDto.setMsg(rpcException.getMessage());
             dispatchResponseDto.setCode(JobResultEnum.DISPATCH_FAIL.getRes());
         }

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java

@@ -14,6 +14,7 @@ import com.byit.thread.BaseThreadRunHelper;
 import com.byit.util.SpringUtil;
 import io.netty.util.TimerTask;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Component;
 
 import javax.sql.DataSource;
@@ -95,16 +96,8 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
 
     private Integer saveLog(JobTaskSchedule mythJobTaskSchedule){
         JobTaskRunLogWithBLOBs jobTaskRunLog = new JobTaskRunLogWithBLOBs();
-
-        jobTaskRunLog.setRunId(mythJobTaskSchedule.getRunId());
-        jobTaskRunLog.setIsVirtual(mythJobTaskSchedule.getIsVirtual());
-        jobTaskRunLog.setFlowId(mythJobTaskSchedule.getFlowId());
-        jobTaskRunLog.setFlowName(mythJobTaskSchedule.getFlowName());
-        jobTaskRunLog.setNodeId(mythJobTaskSchedule.getNodeId());
-        jobTaskRunLog.setNodeName(mythJobTaskSchedule.getNodeName());
-        jobTaskRunLog.setRunParams(mythJobTaskSchedule.getRunParam());
-        jobTaskRunLog.setFailedRemainingCount(mythJobTaskSchedule.getFailedRetryCount());
-        jobTaskRunLog.setJobType(mythJobTaskSchedule.getJobType());
+        BeanUtils.copyProperties(mythJobTaskSchedule, jobTaskRunLog);
+        jobTaskRunLog.setLogId(null);
         JobTaskRunLogServiceImpl mythJobTaskRunLogService = SpringUtil.getBean(JobTaskRunLogServiceImpl.class);
         mythJobTaskRunLogService.saveJobTaskRunLog(jobTaskRunLog);
 

byit-myth-executor/myth-executor-server/src/main/java/com/byit/process/MythJobProcess.java

@@ -53,15 +53,15 @@ public class MythJobProcess implements Callable<String>{
     private String effectiveUser = null;
     //本次任务对应的日志id
     private Integer logId;
-    //redis的发布key
-    private String runId;
     //执行状态，默认是失败
     private int exitCode = 1;
     //redis模板
     private StringRedisTemplate stringRedisTemplate;
+    //redis的key
+    private String runKey;
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
-                          final String workingDir, final Integer logId, final String runId, final StringRedisTemplate stringRedisTemplate) {
+                          final String workingDir, final Integer logId, final StringRedisTemplate stringRedisTemplate) {
         this.cmd = cmd;
         this.env = env;
         this.workingDir = workingDir;
@@ -69,14 +69,14 @@ public class MythJobProcess implements Callable<String>{
         this.startupLatch = new CountDownLatch(1);
         this.completeLatch = new CountDownLatch(1);
         this.logId = logId;
-        this.runId = runId;
         this.stringRedisTemplate = stringRedisTemplate;
+        this.runKey = "REAL-EXEC-" + logId;
     }
 
     public MythJobProcess(final List<String> cmd, final Map<String, String> env,
                           final String workingDir, final String executeAsUserBinary,
                           final String effectiveUser, Integer logId, final String runId, final StringRedisTemplate stringRedisTemplate) {
-        this(cmd, env, workingDir, logId, runId, stringRedisTemplate);
+        this(cmd, env, workingDir, logId, stringRedisTemplate);
         this.isExecuteAsUser = true;
         this.executeAsUserBinary = executeAsUserBinary;
         this.effectiveUser = effectiveUser;
@@ -121,8 +121,8 @@ public class MythJobProcess implements Callable<String>{
                 log.info("进程的id " + this.processId);
             }
 
-            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30, runId, stringRedisTemplate);
-            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30, runId, stringRedisTemplate);
+            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), "utf-8"), Boolean.FALSE, 30, runKey, stringRedisTemplate);
+            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), "utf-8"), Boolean.TRUE, 30, runKey, stringRedisTemplate);
             //开始获取进程的运行日志
             outputGobbler.start();
             errorGobbler.start();
@@ -154,7 +154,7 @@ public class MythJobProcess implements Callable<String>{
                 this.process.destroy();
             }
             RunLog runLog = RunLog.builder().isEnd(true).isSuccess(exitCode == 0 ? true : false).build();
-            stringRedisTemplate.convertAndSend(runId, JSON.toJSONString(runLog, WriteClassName));
+            stringRedisTemplate.convertAndSend(runKey, JSON.toJSONString(runLog, WriteClassName));
             return  outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
         }
     }

byit-myth-executor/myth-executor-server/src/main/java/com/byit/utils/LogGobbler.java

@@ -22,14 +22,14 @@ public class LogGobbler extends Thread {
     private final BufferedReader inputReader;
     private final boolean isError;
     private final StringBuilder buffer;
-    private final String runId;
+    private final String runKey;
     private final StringRedisTemplate stringRedisTemplate;
 
-    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines, final String runId, final StringRedisTemplate stringRedisTemplate) {
+    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines, final String runKey, final StringRedisTemplate stringRedisTemplate) {
         this.inputReader = new BufferedReader(inputReader);
         this.isError = isError;
         this.buffer = new StringBuilder(bufferLines);
-        this.runId = runId;
+        this.runKey = runKey;
         this.stringRedisTemplate = stringRedisTemplate;
     }
 
@@ -44,7 +44,7 @@ public class LogGobbler extends Thread {
                 }
                 this.buffer.append(line + "\n");
                 runLog.setRunLog(line);
-                stringRedisTemplate.convertAndSend(runId, JSON.toJSONString(runLog, WriteClassName));
+                stringRedisTemplate.convertAndSend(runKey, JSON.toJSONString(runLog, WriteClassName));
                 printLog(line);
             }
         } catch (final IOException e) {