runId生成策略修改以及异常修改

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

@@ -20,12 +20,14 @@ import com.byit.model.*;
 import com.byit.model.vo.RunRecordingVo;
 import com.byit.service.ApiFlowService;
 import com.byit.util.ApiFlowDagCheck;
+import com.byit.util.IDGenerationStrategy;
 import com.byit.util.lock.RedissLockUtil;
 import com.byit.utils.ValidationUtil;
 import com.google.common.base.Joiner;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -49,6 +51,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
     private final static SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
     public static final String PRE = "${";
     public static final String SUFFER = "}";
+
+    @Value("${server.port}")
+    private Integer serverPort;
     @Resource
     private FlowMapper flowMapper;
 
@@ -492,7 +497,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         }
 
         //获取当前时间
-        String reRunId = UUID.randomUUID().toString().replace("-","");
+        String reRunId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
         Long triggerTime = System.currentTimeMillis();
         WaitingTask waitingTask = new WaitingTask();
         BeanUtils.copyProperties(jobTaskRunLog, waitingTask);
@@ -598,7 +603,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         }else {
             //不是内嵌工作流
             Long triggerTime = System.currentTimeMillis();
-            String reRunId = UUID.randomUUID().toString().replace("-","");
+            String reRunId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
 
             List<JobTaskRunLogWithBLOBs> jobTaskRunLogList = jobTaskRunLogMapper.findJobTaskRunLogWithBLOBsByFlowIdAndRunId(flow.getFlowId(), runInfo.getRunId());
             ValidationUtil.dataNotNull(jobTaskRunLogList, "该工作流没有在调度上的任务");
@@ -1227,7 +1232,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             for (String repairTime : repairTimeList){
                 WaitingRecord waitingRecord = new WaitingRecord();
                 RunRecording runRecording = new RunRecording();
-                String runId = UUID.randomUUID().toString().replace("-","");
+                String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
                 //设置排期
                 BeanUtils.copyProperties(flow, waitingRecord);
                 waitingRecord.setFlowVersionName(flow.getVersionName());

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

@@ -8,9 +8,11 @@ import com.byit.model.JobTask;
 import com.byit.model.Node;
 import com.byit.model.RunRecording;
 import com.byit.service.*;
+import com.byit.util.IDGenerationStrategy;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.stereotype.Service;
@@ -34,6 +36,8 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
     private final JobTaskService jobTaskService;
     private final FlowService flowService;
     private ApplicationEventPublisher applicationEventPublisher;
+    @Value("${server.port}")
+    private Integer serverPort;
     /**
      * 节点依赖查询操作
      */
@@ -55,7 +59,7 @@ public class RunNodeServiceImpl implements RunNodeServer, ApplicationEventPublis
     @Override
     public void saveRunRec(Flow flow, List<Node> nodes) {
         log.info("---------saveRunRecAndTask start------【保存工作流：{}和节点：{}】-----------------------",flow,nodes);
-        String runId = UUID.randomUUID().toString().replace("-","");
+        String runId = IDGenerationStrategy.runIdGenerationStrategy(serverPort);
         log.info("-------------【开始保存运行记录runId为：{}】------------------",runId);
         RunRecording build = new RunRecording();
         BeanUtils.copyProperties(flow,build);

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

 package com.byit.service.mapservice.impl;
 
+import cn.hutool.core.collection.CollectionUtil;
 import com.byit.enums.EmailEnum;
 import com.byit.enums.FlowPropertyEnum;
 import com.byit.enums.RunRecordingEnum;
@@ -16,6 +17,7 @@ import org.springframework.stereotype.Service;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -56,8 +58,9 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
         //List<NodeVersion> notLogNode = allByFlowId.stream().filter(nodeVersion -> !(logNodeId.contains(nodeVersion.getNodeId()))).collect(Collectors.toList());
         //将这些节点置为失败并将实例也置为失败
         //保存错误日志节点
-
-        //查询task所有的对应节点
+        //第一将非虚节点的节点保存进日志，同时保存虚节点，同时筛选出end节点  不作处理，让他自己执行队形的end节点（为了执行结束事件）
+        List<JobTask> virtualTasks = new ArrayList<>();
+        //查询task所有的对应节点  这里需要判断不是虚节点  在进行日志插入之后需要将对应的节点删除
         List<JobTask> byRunId = jobTaskService.findByRunId(runId);
         byRunId.forEach(task ->{
             JobTaskRunLogWithBLOBs log = new JobTaskRunLogWithBLOBs();
@@ -73,6 +76,11 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
             log.setFlowName(runRecording.getFlowName());
             log.setTriggerTime(thisTime);
             jobTaskRunLogService.saveJobTaskRunLog(log);
+
+            if ("0".equals(task.getIsVirtual())) {
+                virtualTasks.add(task);
+            }
+            jobTaskService.removeMythJobTaskById(task.getId());
         });
         //执行实例的快速失败
         if (runRecording.getStartTime() == null) {
@@ -85,8 +93,6 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
         runRecordingService.updateRunRecordingById(runRecording);
         applicationEventPublisher.publishEvent(new EndFlowEvent(this,runRecording.getFlowId()));
 
-        //筛选虚节点
-        List<JobTask> virtualTasks = byRunId.stream().filter(task -> "0".equals(task.getIsVirtual())).collect(Collectors.toList());
         virtualTasks.forEach(virtualTask -> {
             Integer mapFlowId = virtualTask.getMapFlowId();
             List<NodeVersion> allByFlowId = nodeVersionService.findAllByFlowId(mapFlowId);
@@ -144,8 +150,7 @@ public class RunRecordingAndLogServiceImpl implements RunRecordingAndLogService,
         });
 
         //删除task里面的数据
-        jobTaskService.removeByRunId(runId);
-
+        //jobTaskService.removeByRunId(runId);
 
     }
 

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

@@ -113,7 +113,12 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                             e.printStackTrace();
                         }
                     } catch (Exception e) {
-                        e.printStackTrace();
+                        log.error("-----------------节点执行时出现异常,{}----------",e.getMessage());
+                        try {
+                            taskAndLogServer.addRunLogAndRemoveTask(jobTask,true,true);
+                        } catch (UnknownHostException unknownHostException) {
+                            log.error("-----------------节点执行时出现异常,{}----------",unknownHostException.getMessage());
+                        }
                     }
                 } else {
                     //先处理开始节点
@@ -138,7 +143,7 @@ public class JobTaskThreadRunHelper extends BaseThreadRunHelper {
                             } catch (UnknownHostException e) {
                                 log.error("-----------------节点执行时出现异常,{}----------",e.getMessage());
                             }
-                        }catch (BusinessException businessException) {
+                        }catch (Exception businessException) {
                             try {
                                 taskAndLogServer.addRunLogAndRemoveTask(jobTask,false,true);
                             } catch (UnknownHostException e) {

byit-myth-core/myth-admin-core/src/main/java/com/byit/util/IDGenerationStrategy.java

+package com.byit.util;
+
+import com.byit.utils.IpUtil;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.UUID;
+
+/**
+ *
+ * @author huangfu
+ */
+public class IDGenerationStrategy {
+
+    public static String runIdGenerationStrategy(int port){
+        String runIdPre = UUID.randomUUID().toString().replace("-","");
+        String ipPort = IpUtil.getIpPort(port);
+        if (StringUtils.isNoneBlank(ipPort)){
+            String replace = ipPort.replace(".", "").replace(":", "");
+            runIdPre = runIdPre + replace;
+        }
+
+        return runIdPre;
+    }
+
+    public static void main(String[] args) {
+        System.out.println(runIdGenerationStrategy(88));
+
+    }
+}