增加jobUtil错误重试，增加立即运行节点强制超时判断

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/JobTaskRunLogMapper.java

@@ -19,7 +19,7 @@ public interface JobTaskRunLogMapper {
   * 查询立即运行节点未完成的
   * @return 所有运行中的
   */
-  List<JobTaskRunLog> findTempRunNodeList();
+  List<JobTaskRunLogWithBLOBs> findTempRunNodeList();
 
   List<JobTaskRunLogWithBLOBs> findAllByType(@Param("type") Integer type,@Param("jobName") String jobName);
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/JobTaskRunLogService.java

@@ -18,7 +18,7 @@ public interface JobTaskRunLogService {
      * 查询立即运行节点未完成的
      * @return 所有运行中的
      */
-    List<JobTaskRunLog> findTempRunNodeList();
+    List<JobTaskRunLogWithBLOBs> findTempRunNodeList();
 
     List<JobTaskRunLogWithBLOBs> findAllByType(Integer type,String jobName);
 

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

@@ -41,7 +41,7 @@ public class JobTaskRunLogServiceImpl implements JobTaskRunLogService {
     }
 
     @Override
-    public List<JobTaskRunLog> findTempRunNodeList() {
+    public List<JobTaskRunLogWithBLOBs> findTempRunNodeList() {
         return jobTaskRunLogMapper.findTempRunNodeList();
     }
 

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

@@ -3,7 +3,9 @@ package com.byit.thread.helper;
 import cn.hutool.core.collection.CollectionUtil;
 import com.alibaba.fastjson.JSON;
 import com.byit.dto.plugin.RunLog;
+import com.byit.enums.task.RunResultEnum;
 import com.byit.model.JobTaskRunLog;
+import com.byit.model.JobTaskRunLogWithBLOBs;
 import com.byit.service.JobTaskRunLogService;
 import com.byit.thread.BaseThreadRunHelper;
 import com.byit.util.IDGenerationStrategy;
@@ -12,10 +14,12 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
 
 /**
+ * 立即运行节点执行监控 24小内的节点信息   24小时内未执行完的数据全部置为失败，删除redis的心跳信息
  * @author huangfu
  */
 @Slf4j
@@ -26,6 +30,10 @@ public class UnfinishedNodeMonitoringRunHelper extends BaseThreadRunHelper {
      */
     private static final String PENG = "PENG";
     /**
+     * 默认一天
+     */
+    private static final Long OVERTIME_TIME = TimeUnit.DAYS.toMillis(1);
+    /**
      * 心跳前缀
      */
     private final StringRedisTemplate stringRedisTemplate;
@@ -38,13 +46,23 @@ public class UnfinishedNodeMonitoringRunHelper extends BaseThreadRunHelper {
 
     @Override
     public Long start() {
-        List<JobTaskRunLog> tempRunNodeList = jobTaskRunLogService.findTempRunNodeList();
+        List<JobTaskRunLogWithBLOBs> tempRunNodeList = jobTaskRunLogService.findTempRunNodeList();
         if (CollectionUtil.isNotEmpty(tempRunNodeList)) {
             tempRunNodeList.forEach(nodeLog -> {
-                log.debug("-----节点{}未执行完毕-----", nodeLog);
                 String keyGenerationStrategy = IDGenerationStrategy.keyGenerationStrategy(nodeLog.getLogId());
-                RunLog runLog = RunLog.builder().isEnd(false).runLog(PENG).build();
-                stringRedisTemplate.opsForList().rightPush(keyGenerationStrategy, JSON.toJSONString(runLog, WriteClassName));
+                if(System.currentTimeMillis() - nodeLog.getTriggerTime().getTime() > OVERTIME_TIME){
+                    log.warn("-----节点{}，执行时间超时，强制失败，删除心跳节点-----",nodeLog);
+                    nodeLog.setRunCode(RunResultEnum.RUN_ERROR.getCode());
+                    nodeLog.setRunMsg("节点执行超时，强制失败！");
+                    stringRedisTemplate.delete(keyGenerationStrategy);
+                    jobTaskRunLogService.updateJobTaskRunLog(nodeLog);
+                }else{
+                    log.debug("-----节点{}未执行完毕-----", nodeLog);
+                    RunLog runLog = RunLog.builder().isEnd(false).runLog(PENG).build();
+                    stringRedisTemplate.opsForList().rightPush(keyGenerationStrategy, JSON.toJSONString(runLog, WriteClassName));
+                }
+
+
             });
         }
         return 5000L;

byit-myth-core/myth-admin-core/src/main/resources/mapper/JobTaskRunLogMapper.xml

@@ -49,8 +49,12 @@
     run_msg, trigger_msg
   </sql>
 
-  <select id="findTempRunNodeList" resultMap="BaseResultMap">
-    select <include refid="Base_Column_List" /> from job_task_run_log where schedule_type = 4 and run_code = '0'
+  <select id="findTempRunNodeList" resultMap="ResultMapWithBLOBs">
+    select
+        <include refid="Base_Column_List" />
+        ,
+        <include refid="Blob_Column_List" />
+    from job_task_run_log where schedule_type = 4 and run_code = '0'
   </select>
 
   <!--根据runId 和 上级节点的集合 查询所有的上级节点-->

byit-myth-executor/myth-executor-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -8,6 +8,7 @@ import com.byit.dto.plugin.*;
 import com.byit.dto.specials.SpecialJobParam;
 import com.byit.dto.web.ResponseResult;
 import com.byit.executor.handler.interfaces.IJobHandler;
+import com.byit.rpc.util.RPCLogUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 
@@ -29,6 +30,10 @@ import static com.alibaba.fastjson.serializer.SerializerFeature.WriteClassName;
 public class JobUtils {
     private static String REQUEST_URL;
     /**
+     * 错误重试次数
+     */
+    private static Integer DEFAULT_RETRY_COUNT = 3;
+    /**
      * 存储token
      */
     private static final ThreadLocal<String> TOKEN_THREADLOCAL = new ThreadLocal<String>();
@@ -217,13 +222,44 @@ public class JobUtils {
         if (StringUtils.isBlank(token)) {
             throw new RuntimeException("请求的token尚未设置！");
         }
-        HttpRequest httpRequest = HttpRequest.post(REQUEST_URL + requestUrl);
-        httpRequest.header("token", token);
-        String response = httpRequest.body(body).execute().body();
+
+        String response = requestPostServer(requestUrl, body, token, DEFAULT_RETRY_COUNT, 0);
         return response;
     }
 
     /**
+     * 重试
+     * @param requestUrl 请求的url
+     * @param body 数据包
+     * @param retryTotalCount 总共的重试次数
+     * @param thisRetryCount 当前的重试次数
+     * @return
+     */
+    private static String requestPostServer(String requestUrl, String body, String token ,int retryTotalCount, int thisRetryCount) {
+        log.info("------当前的请求的地址为:{}-------",requestUrl);
+        try {
+            HttpRequest httpRequest = HttpRequest.post(REQUEST_URL + requestUrl);
+            httpRequest.header("token", token);
+            String response = httpRequest.body(body).execute().body();
+            log.info("------{}http通道建立成功，重试了{}次------",requestUrl,thisRetryCount-1);
+            return response;
+        }catch (Exception e) {
+            //当前重试次数 小于等于总共的重试次数时
+            if(thisRetryCount <= retryTotalCount){
+                log.error("{},http通道建立时出现异常，异常信息为：{}，开始第{}次重试！",requestUrl, RPCLogUtil.getMessage(e), thisRetryCount);
+                try {
+                    Thread.sleep(1000 * thisRetryCount);
+                } catch (InterruptedException interruptedException) {
+                    interruptedException.printStackTrace();
+                }
+                requestPostServer(requestUrl,body,token,retryTotalCount,++thisRetryCount);
+            }
+            log.error("----与主机【{}】建立通道，总共【{}】次，全部失败",retryTotalCount, retryTotalCount);
+        }
+        return null;
+    }
+
+    /**
      * 特殊的补批接口
      *
      * @return