修改扫描线程为线程池运行

byit-myth-admin/src/main/java/com/byit/controller/EmailController.java

 package com.byit.controller;
 
 import com.byit.exception.DataValidationException;
+import com.byit.factory.DaemonScanThreadRunHelperRedisLock;
 import com.byit.model.vo.EmailAlarmVo;
 import com.byit.service.EmailAlarmService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -17,22 +18,9 @@ import java.util.UUID;
  **/
 @RestController
 public class EmailController {
-    private final EmailAlarmService emailAlarmService;
 
-    @Autowired
-    public EmailController(EmailAlarmService emailAlarmService) {
-        this.emailAlarmService = emailAlarmService;
-    }
-
-    @RequestMapping("send")
-    public String send(){
-        EmailAlarmVo emailAlarmVo = new EmailAlarmVo();
-        emailAlarmVo.setFlowId(1);
-        emailAlarmVo.setFlowName("测试工作流");
-        emailAlarmVo.setRunId(UUID.randomUUID( ).toString());
-        emailAlarmVo.setVersionName("V1");
-        emailAlarmVo.setAlarmEmail("huangfukexing@byitgroup.com");
-        emailAlarmService.sendEmail(emailAlarmVo);
-        return "success";
+    @RequestMapping("/threadCodeCount")
+    public int getAcCount(){
+        return DaemonScanThreadRunHelperRedisLock.SCAN_WORLD_THREAD.getActiveCount();
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelperRedisLock.java

@@ -2,6 +2,7 @@ package com.byit.factory;
 
 import com.byit.thread.BaseThreadRunHelper;
 import com.byit.job.utils.MythLogUtils;
+import com.byit.util.ThreadPoolUtil;
 import com.byit.util.lock.RedissLockUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -9,6 +10,7 @@ import org.springframework.stereotype.Component;
 
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -20,6 +22,11 @@ import java.util.concurrent.TimeUnit;
 @ConditionalOnExpression("'${lock.type}'.equals('redis')")
 public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunHelper {
 
+    /**
+     * 扫描线程数量
+     */
+    public static ThreadPoolExecutor SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor("scan flow stream thread",THREAD_RUN_OPT.entrySet().size());
+
     @Override
     public void runDaemonThreads() {
         Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
@@ -62,7 +69,8 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
         });
         exampleThread.setName(threadName);
         exampleThread.setDaemon(true);
-        exampleThread.start();
+        SCAN_WORLD_THREAD.execute(exampleThread);
+        //exampleThread.start()
         THREADS_MAP.put(threadName,exampleThread);
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/flowservice/impl/NodeVerificationImpl.java

@@ -64,7 +64,7 @@ public class NodeVerificationImpl implements NodeVerification {
         String nodeDepend = thisJobTask.getNodeDepend();
         if(StringUtils.isBlank(nodeDepend)){
             //对于这个操作，外部捕获到这个异常后应该将该节点写入日志，并设置异常信息
-            throw new BusinessException(NodeRunStatusPropertyEnum.NODE_RELY_ERROR.getMsg());
+            return true;
         }
         String[] split = nodeDepend.split(NODEDEPEND_SPLIT);
         //根据依赖节点查询对应的日志信息

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

@@ -14,7 +14,8 @@ public interface TaskAndLogServer {
      * @param jobTask 任务节点
      * @param isInner 虚节点
      * @param taskError 节点本身是否异常
+     * @param errorMsg 错误信息
      * @exception UnknownHostException ddd
      */
-    void addRunLogAndRemoveTask(JobTask jobTask,boolean isInner,boolean taskError) throws UnknownHostException;
+    void addRunLogAndRemoveTask(JobTask jobTask,boolean isInner,boolean taskError,String... errorMsg) throws UnknownHostException;
 }

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

@@ -232,6 +232,8 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
     public void updateRunRecordingAndTask(WaitingRecord waitingRecord) {
         log.debug("---------开始查询等待工作流{}对应的数据-------------",waitingRecord);
         RunRecording runRecording = runRecordingService.findAllByRunID(waitingRecord.getRunId());
+
+        runRecording.setStartTime(new Date());
         //将运行实例改为以执行
         runRecording.setFlowStatus(RunRecordingEnum.FLOW_STATUS_RUN_ING.getCode());
         runRecordingService.updateRunRecordingById(runRecording);

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

@@ -49,7 +49,7 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
     }
 
     @Override
-    public void addRunLogAndRemoveTask(JobTask jobTask, boolean isInner,boolean taskError) throws UnknownHostException {
+    public void addRunLogAndRemoveTask(JobTask jobTask, boolean isInner,boolean taskError,String... errorMsg) throws UnknownHostException {
         Date thisDate = new Date();
         //删除任务节点
         jobTaskService.removeMythJobTaskById(jobTask.getId());
@@ -76,7 +76,7 @@ public class TaskAndLogServerImpl implements TaskAndLogServer {
         jobTaskRunLog.setMapFlowId(jobTask.getMapFlowId());
         if (taskError) {
             jobTaskRunLog.setRunCode("2");
-            jobTaskRunLog.setRunMsg("节点执行时异常");
+            jobTaskRunLog.setRunMsg(StringUtils.join(errorMsg,","));
         }else {
             jobTaskRunLog.setRunCode("6");
             jobTaskRunLog.setRunMsg("上级节点执行失败");

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

+package com.byit.util;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.util.concurrent.*;
+
+/**
+ * 线程池配置
+ * @author huangfu
+ */
+public class ThreadPoolUtil {
+
+    /**
+     * 创建线程池。特性如下：
+     * core大小为8，初始状态下8个线程，无初始消耗。
+     * max大小为16，最多16个线程。
+     * 60秒超时时间，闲置超过60秒线程会被回收。
+     * 使用SynchronousQueue，任务不会排队，必须要有可用线程才能提交成功，否则会RejectedExecutionException。
+     * @return 线程池
+     */
+    public static ThreadPoolExecutor createThreadPoolExecutor(String threadName) {
+        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("[byit-myth-job]-"+threadName+"-Thread-%d").build();
+        return new ThreadPoolExecutor(8, 16,
+                60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory,  new ThreadPoolExecutor.AbortPolicy());
+    }
+
+
+    /**
+     * 创建定长线程池。特性如下： customize
+     * @param threadName 线程名称
+     * @param threadCount 线程数量
+     * @return 线程池
+     */
+    public static ThreadPoolExecutor  createFixedLengthThreadPoolExecutor(String threadName, int threadCount) {
+        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("[byit-myth-job]-"+threadName+"-Thread-%d").build();
+        return new ThreadPoolExecutor(threadCount, threadCount,
+                0L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory,  new ThreadPoolExecutor.AbortPolicy());
+    }
+
+
+    /**
+     * 定制化线程池
+     * @param threadName 线程名称
+     * @param coreCount 核心线程数量
+     * @param maxCount 最大线程数量
+     * @param keepAliveTime 线程保持时间
+     * @param timeUnit 时间单位
+     * @param queueLength 队列长度
+     * @return 返回对应的线程池
+     */
+    public static ThreadPoolExecutor  createCutomizeThreadPoolExecutor(String threadName, int coreCount, int maxCount,
+                                                                       long keepAliveTime, TimeUnit timeUnit, int queueLength) {
+        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("[byit-myth-job]-"+threadName+"-Thread-%d").build();
+        return new ThreadPoolExecutor(coreCount, maxCount,
+                keepAliveTime, timeUnit,
+                new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory,  new ThreadPoolExecutor.AbortPolicy());
+    }
+}