修复补批工作流执行重复的BUG

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

@@ -42,7 +42,7 @@ public class RunRecordingServiceImpl implements RunRecordingService {
     /**
      * 是否存在运行的工作流实例
      * @param flowId 工作流ID
-     * @return 是否在正在运行中
+     * @return true有正在运行中的数据   反之没有
      */
     @Override
     public boolean findRunRecordingIsRunning(Integer flowId) {

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

 package com.byit.thread.helper;
 
+import com.byit.model.RunRecording;
 import com.byit.model.WaitingRecord;
+import com.byit.service.RunRecordingService;
 import com.byit.service.WaitingRecordService;
 import com.byit.service.mapservice.RunRecordingAndJobTaskService;
 import com.byit.thread.BaseThreadRunHelper;
@@ -28,14 +30,16 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
 
     private final WaitingRecordService waitingRecordService;
     private final RunRecordingAndJobTaskService runRecordingAndJobTaskService;
+    private final RunRecordingService runRecordingService;
 
 
 
 
-    public MakeUpFlowThreadRunHelper(DataSource dataSource, WaitingRecordService waitingRecordService, RunRecordingAndJobTaskService runRecordingAndJobTaskService) {
+    public MakeUpFlowThreadRunHelper(DataSource dataSource, WaitingRecordService waitingRecordService, RunRecordingAndJobTaskService runRecordingAndJobTaskService, RunRecordingService runRecordingService) {
         this.dataSource = dataSource;
         this.waitingRecordService = waitingRecordService;
         this.runRecordingAndJobTaskService = runRecordingAndJobTaskService;
+        this.runRecordingService = runRecordingService;
     }
 
     @Override
@@ -51,8 +55,12 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
         Map<Integer, WaitingRecord> nextRunFlow = allByTriggerTime.stream()
                 .collect(Collectors.toMap(WaitingRecord::getFlowId, Function.identity(), BinaryOperator.minBy(Comparator.comparingInt(WaitingRecord::getWaitOrder))));
 
+
         nextRunFlow.forEach((key,value) ->{
-            runRecordingAndJobTaskService.updateRunRecordingAndTask(value);
+            boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(value.getFlowId());
+            if(!runRecordingIsRunning){
+                runRecordingAndJobTaskService.updateRunRecordingAndTask(value);
+            }
         });
 
         return UNIVERSAL_WAIT_TIME;