多实例支持

byit-myth-admin/src/main/java/com/byit/api/ApiFlowController.java

@@ -208,11 +208,4 @@ public class ApiFlowController {
         return ResponseResult.ok(collectData);
     }
 
-    @PostMapping("/loadCurrentStatus")
-    @ApiOperation("获取当前的工作流运行状态")
-    public ResponseResult loadCurrentStatus(String param) {
-        Map<String, RunRecording> statusMap = apiFlowService.loadCurrentStatus(param);
-        return ResponseResult.ok(statusMap);
-    }
-
 }

byit-myth-admin/src/main/java/com/byit/service/ApiFlowService.java

@@ -99,12 +99,6 @@ public interface ApiFlowService {
      */
     CollectData loadNodeStatisticData(String param);
 
-    /**
-     * 获取工作流的当前状态
-     * @param param
-     * @return
-     */
-    Map<String, RunRecording> loadCurrentStatus(String param);
 
     /**
      * 功能描述 判断工作流是否存在

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

@@ -131,7 +131,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         Workspace workspace = validate(pluginPackage);
         //保存工作流
         log.debug("保存工作流和节点信息");
-        Flow flow = saveFlow(pluginPackage.getFlow(), workspace.getWorkspaceId(), false, "");
+        Flow flow = saveFlow(pluginPackage.getFlow(), workspace.getWorkspaceId(), false, "",null);
         //生成工作流版本
         log.debug("生成版本");
         FlowVersion flowVersion = saveFlowVersion(flow);
@@ -148,7 +148,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
      * @param isInnner
      * @param workspaceId
      */
-    private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner, String upFlowName) throws Exception {
+    private Flow saveFlow(PluginFlow pluginFlow, Integer workspaceId, boolean isInnner, String upFlowName, String hasManyInstance) throws Exception {
         //拼装工作流名称
         upFlowName = String.format("%s:%s", upFlowName, pluginFlow.getName());
 
@@ -157,6 +157,16 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         if (extendedConfiguration == null) {
             extendedConfiguration = new FlowExtendedConfiguration();
         }
+
+        //当自身的工作流上携带是否多实例的参数的时候，使用自己的, 如果为空的话就使用父类的
+        if(StringUtils.isBlank(extendedConfiguration.getHasManyInstance())) {
+            if(StringUtils.isNoneBlank(hasManyInstance)){
+                extendedConfiguration.setHasManyInstance(hasManyInstance);
+            }else{
+                extendedConfiguration.setHasManyInstance(FlowPropertyEnum.NOT_MANY_INSTANCE.getCode());
+            }
+        }
+        hasManyInstance = extendedConfiguration.getHasManyInstance();
         extendedConfiguration.setFlowEmbedLogo(upFlowName);
         flow.setExtendedConfiguration(JSON.toJSONString(extendedConfiguration));
         flow.setFlowName(pluginFlow.getName());
@@ -206,7 +216,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
                 flow.setVersionName(versionName);
             }
             flowMapper.updateByIdSelective(flow);
-            updateNodeByFlow(pluginFlow.getNodeList(), flow, upFlowName);
+            updateNodeByFlow(pluginFlow.getNodeList(), flow, upFlowName,hasManyInstance);
 
         } else {
             if (StringUtils.isBlank(versionName)) {
@@ -216,7 +226,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             }
 
             Integer flowId = flowMapper.insertSelective(flow);
-            saveNode(pluginFlow.getNodeList(), flow, upFlowName);
+            saveNode(pluginFlow.getNodeList(), flow, upFlowName, hasManyInstance);
         }
         return flow;
     }
@@ -227,7 +237,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
      * @param pluginNodeList
      * @param flow
      */
-    private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow, String upFlowName) throws Exception {
+    private void updateNodeByFlow(List<PluginBaseNode> pluginNodeList, Flow flow, String upFlowName,String hasManyInstance) throws Exception {
 
         //将工作流下的所有节点设为不在工作流调度上
         log.debug("将工作流【{}】所有的节点移下调度,并删除相关的依赖关系", flow.getFlowName());
@@ -257,7 +267,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         //保存节点信息
         for (PluginBaseNode pluginNode : pluginNodeList) {
             Node node = new Node();
-            buildNode(pluginNode, flow, node, upFlowName);
+            buildNode(pluginNode, flow, node, upFlowName,hasManyInstance);
 
             Integer nodeId = nodeMapper.insertSelective(node);
             nameIdRel.put(node.getNodeName(), node.getNodeId());
@@ -272,12 +282,12 @@ public class ApiFlowServiceImpl implements ApiFlowService {
      * @param nodeList
      * @param flow
      */
-    private void saveNode(List<PluginBaseNode> nodeList, Flow flow, String upFlowName) throws Exception {
+    private void saveNode(List<PluginBaseNode> nodeList, Flow flow, String upFlowName, String hasManyInstance) throws Exception {
         HashMap<String, Integer> nameIdRel = new HashMap<>();
         //保存节点信息
         for (PluginBaseNode pluginNode : nodeList) {
             Node node = new Node();
-            buildNode(pluginNode, flow, node, upFlowName);
+            buildNode(pluginNode, flow, node, upFlowName, hasManyInstance);
             node.setNodeId(null);
             Integer nodeId = nodeMapper.insertSelective(node);
             nameIdRel.put(node.getNodeName(), node.getNodeId());
@@ -293,7 +303,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
      * @param node
      * @throws Exception
      */
-    private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node, String upFlowName) throws Exception {
+    private void buildNode(PluginBaseNode pluginNode, Flow flow, Node node, String upFlowName, String hasManyInstance) throws Exception {
 
         node.setVersionName(flow.getVersionName());
         node.setFlowId(flow.getFlowId());
@@ -306,7 +316,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             PluginFlow pluginFlow = (PluginFlow) pluginNode;
             pluginFlow.getConfig().setFlowCron(flow.getFlowCron());
             //如果是内嵌工作流先保存工作流信息
-            Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true, upFlowName);
+            Flow innerFlow = saveFlow(pluginFlow, flow.getWorkspaceId(), true, upFlowName, hasManyInstance);
             node.setIsVirtual(NodePropertyEnum.IS_VIRTUAL.getCode());
             node.setMapFlowId(innerFlow.getFlowId());
             node.setVersionName(pluginFlow.getVersionName());
@@ -1171,35 +1181,6 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         return buildStatisticAllDate(startTime, endTime, workspace.getWorkspaceId(), flowName);
     }
 
-    @Override
-    public Map<String, RunRecording> loadCurrentStatus(String param) {
-        ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
-        JSONObject jsonObject = JSON.parseObject(param);
-        //获取工作空间名称
-        String workspaceName = jsonObject.getString("workspaceName");
-        ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空！");
-        Workspace workspace = workspaceMapper.getByName(workspaceName);
-        ValidationUtil.dataNotNull(workspace, workspaceName + "工作空间不存在");
-        //获取工作流名称
-        String flowNames = jsonObject.getString("flowNames");
-        ValidationUtil.dataNotBank(flowNames, "工作流名称不允许为空！");
-        List<String> flowNameList = Arrays.asList(flowNames.split(","));
-        List<Flow> flowList = new ArrayList<>();
-        flowNameList.forEach(flowName -> {
-            Flow flow = flowMapper.getByWorkSpaceAndName(workspace.getWorkspaceId(), flowName);
-            ValidationUtil.dataNotNull(flow, flowName + "工作流不存在");
-            flowList.add(flow);
-        });
-
-        Map<String, RunRecording> statusMap = new HashMap<>();
-
-        flowList.forEach(flow -> {
-            RunRecording runRecording = runRecordingMapper.findNewStatus(flow.getFlowId());
-            statusMap.put(flow.getFlowName(), runRecording);
-        });
-
-        return statusMap;
-    }
 
     @Override
     public Boolean exist(String param) {
@@ -1526,6 +1507,10 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         deleteFlowAndNode(flow);
     }
 
+    /**
+     * TODO 这个是暂停所有实例
+     * @param param
+     */
     @Override
     public void unsuspendFlow(String param) {
         ValidationUtil.dataNotBank(param, "请求参数不允许为空！");
@@ -1538,8 +1523,9 @@ public class ApiFlowServiceImpl implements ApiFlowService {
         ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空！");
         Workspace byName = workspaceMapper.getByName(workspaceName);
         ValidationUtil.isTrueValidation(byName == null, "没有此工作空间");
-        RunRecording runRecording = runRecordingMapper.runIngRunRecording(flowName, byName.getWorkspaceId());
-        ValidationUtil.isTrueValidation(runRecording == null, "没有正在运行中的实例！");
+        List<RunRecording> runRecordingList = runRecordingMapper.runIngRunRecording(flowName, byName.getWorkspaceId());
+        ValidationUtil.isTrueValidation(CollectionUtil.isEmpty(runRecordingList), "没有正在运行中的实例！");
+        runRecordingList.forEach(runRecording -> {
             if ("3".equals(runRecording.getFlowStatus())) {
                 StopFlowParam stopFlowParam = new StopFlowParam();
                 stopFlowParam.setRunIds(runRecording.getRunId());
@@ -1549,6 +1535,8 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             } else {
                 log.warn("{}，不是出于暂停状态", runRecording);
             }
+        });
+
 
     }
 

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

@@ -59,6 +59,8 @@ public class ApiNodeOperatingServiceImpl implements ApiNodeOperatingService {
         //工作空间名称
         String workspaceName = stopNodeParam.getWorkspaceName();
         ValidationUtil.dataNotBank(workspaceName, "工作空间名称不允许为空");
+        String runId = stopNodeParam.getRunId();
+        ValidationUtil.dataNotBank(runId, "运行标识不允许为空");
         //那个工作流
         String flowName = stopNodeParam.getFlowName();
         ValidationUtil.dataNotBank(flowName, "工作流名称不允许为空");
@@ -71,7 +73,7 @@ public class ApiNodeOperatingServiceImpl implements ApiNodeOperatingService {
         Flow flow = flowService.findFlowByName(flowName, workspace.getWorkspaceId());
         ValidationUtil.dataNotNull(flow, "工作流不存在！");
         //运行实例
-        RunRecording runRecording = runRecordingService.runIngRunRecording(flowName, workspace.getWorkspaceId());
+        RunRecording runRecording = runRecordingService.findAllByRunIDAndFlowName(runId, flowName);
         ValidationUtil.dataNotNull(runRecording, "没有运行中的实例！");
         //查询所有的节点
         List<Node> flowAllNode = nodeService.findNodeByFlowIdAndVersionName(flow.getFlowId());

byit-myth-core/myth-admin-core/src/main/java/com/byit/enums/FlowPropertyEnum.java

@@ -23,7 +23,9 @@ public enum FlowPropertyEnum {
     ISNOT_CURRENTVERSION("1", "版本表不是当前版本的工作流"),
     FLOW_RUN_ING("2","工作流运行中"),
     SCAN("1","标识扫描"),
-    NOT_SCAN("2","不允许扫描")
+    NOT_SCAN("2","不允许扫描"),
+    NOT_MANY_INSTANCE("1","不支持多实例"),
+    MANY_INSTANCE("0","支持多实例"),
     ;
 
     private String code;

byit-myth-core/myth-admin-core/src/main/java/com/byit/listener/FlowEventListener.java

 package com.byit.listener;
 
+import com.alibaba.fastjson.JSON;
+import com.byit.dto.plugin.FlowExtendedConfiguration;
 import com.byit.enums.FlowPropertyEnum;
 import com.byit.event.EndFlowEvent;
 import com.byit.event.FlowScanEndEvent;
 import com.byit.model.Flow;
+import com.byit.model.RunRecording;
 import com.byit.service.FlowService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.event.EventListener;
@@ -31,13 +34,21 @@ public class FlowEventListener {
     @EventListener
     public void flowScanEndEventListener(FlowScanEndEvent flowScanEndEvent){
         log.info("-----------监听到事件{}，工作流添加进实例完成事件-------",flowScanEndEvent);
+        RunRecording runRecording = flowScanEndEvent.getRunRecording();
+        String extendedConfiguration = runRecording.getExtendedConfiguration();
+        FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
+        String hasManyInstance = flowExtendedConfiguration.getHasManyInstance();
+        if(FlowPropertyEnum.NOT_MANY_INSTANCE.getCode().equals(hasManyInstance)){
+            //不支持多实例
             Flow flow = Flow.builder()
-                .flowId(flowScanEndEvent.getRunRecording().getFlowId())
+                    .flowId(runRecording.getFlowId())
                     .scanMark(FlowPropertyEnum.NOT_SCAN.getCode())
                     .build();
             flowService.updateByIdSelective(flow);
         }
 
+    }
+
     /**
      * 工作流完成事件监听
      * @param endFlowEvent 事件信息

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

@@ -49,7 +49,7 @@ public interface RunRecordingMapper {
      * @param workspaceId
      * @return
      */
-    RunRecording runIngRunRecording(@Param("flowName") String flowName, @Param("workspaceId") Integer workspaceId);
+    List<RunRecording> runIngRunRecording(@Param("flowName") String flowName, @Param("workspaceId") Integer workspaceId);
     /**
      * 查询全部的数据
      * @return

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

@@ -13,7 +13,6 @@ import java.util.List;
  */
 public interface RunRecordingService {
 
-    RunRecording runIngRunRecording(String flowName,Integer workspaceId);
 
     /**
      * 查询当天的实例  查询一个时间段的范围

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

@@ -41,11 +41,6 @@ public class RunRecordingServiceImpl implements RunRecordingService  {
         this.workspaceService = workspaceService;
     }
 
-    @Override
-    public RunRecording runIngRunRecording(String flowName,Integer workspaceId) {
-        return runRecordingMapper.runIngRunRecording(flowName, workspaceId);
-    }
-
     /**
      * 查询当天的数据
      * @param startTime 开始时间

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

 package com.byit.service.mapservice.impl;
 
 import cn.hutool.core.collection.CollectionUtil;
+import com.alibaba.fastjson.JSON;
 import com.byit.dto.BeanStrategyPackage;
+import com.byit.dto.plugin.FlowExtendedConfiguration;
 import com.byit.enums.*;
 import com.byit.enums.task.RunResultEnum;
 import com.byit.job.utils.CronExpression;
@@ -77,6 +79,17 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
         Flow mainFlow = flowService.findFlowById(jobTask.getFlowId());
         //内嵌工作流
         Flow virFlow = flowService.findFlowById(mapFlowId);
+        //获取该虚节点的扩展配置
+        String extendedConfiguration = virFlow.getExtendedConfiguration();
+        //格式化扩展配置
+        FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
+        String hasManyInstance = flowExtendedConfiguration.getHasManyInstance();
+        //如果虚节点设置为不支持多实例
+        if (FlowPropertyEnum.NOT_MANY_INSTANCE.getCode().equals(hasManyInstance)) {
+            log.info("不支持多实例{}",virFlow);
+            return;
+        }
+
         virFlow.setScanMark("2");
         RunRecording mainRecording = runRecordingService.findRunRecordingByFlowIdAndRunId(jobTask.getFlowId(), jobTask.getRunId());
 
@@ -127,7 +140,7 @@ public class RunRecordingAndJobTaskServiceImpl implements RunRecordingAndJobTask
                 .failFast(RunRecordingEnum.FAIL_FAST_NO.getCode())
                 .workspaceId(virFlow.getWorkspaceId())
                 .repeatTime(mainRecording.getRepeatTime())
-                .extendedConfiguration(virFlow.getExtendedConfiguration())
+                .extendedConfiguration(extendedConfiguration)
                 .build();
         if(repair) {
             runRecording.setTriggerTime(mainRecording.getTriggerTime());

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

@@ -126,7 +126,7 @@ public class JavaFlowFreeTaskCallbackThread implements Runnable {
             return null;
         }
 
-        RunRecording runRecording = bean.runIngRunRecording(flowName, byName.getWorkspaceId());
+        RunRecording runRecording = bean.findAllByRunIDAndFlowName(specialJavaNode.getRunId(), flowName);
         if (runRecording == null) {
             return null;
         }

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

 package com.byit.thread.helper;
 
 import com.alibaba.fastjson.JSON;
+import com.byit.dto.plugin.FlowExtendedConfiguration;
+import com.byit.enums.FlowPropertyEnum;
 import com.byit.model.WaitingRecord;
 import com.byit.service.RunRecordingService;
 import com.byit.service.WaitingRecordService;
@@ -62,10 +64,18 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
         log.debug("-------筛选后的数据为：{}----------", JSON.toJSONString(nextRunFlow));
 
         for (WaitingRecord value : nextRunFlow.values()) {
+            //查找是否有工作流正在运行中  true有  false没有
             boolean runRecordingIsRunning = runRecordingService.findRunRecordingIsRunning(value.getFlowId());
+            //获取扩展对象
+            String extendedConfiguration = value.getExtendedConfiguration();
+            //转换扩展对象
+            FlowExtendedConfiguration flowExtendedConfiguration = JSON.parseObject(extendedConfiguration, FlowExtendedConfiguration.class);
+            //获取是否支持多实例
+            String hasManyInstance = flowExtendedConfiguration.getHasManyInstance();
             log.debug("-------{}的运行状态为{}",value,runRecordingIsRunning);
             log.debug("==================================================的运行状态为{}======================================" ,runRecordingIsRunning);
-            if(!runRecordingIsRunning){
+            //如果支持多实例 直接运行
+            if(!runRecordingIsRunning || FlowPropertyEnum.MANY_INSTANCE.getCode().equals(hasManyInstance)){
                 runRecordingAndJobTaskService.updateRunRecordingAndTask(value);
             }
         }

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/operating/StopNodeParam.java

@@ -24,4 +24,6 @@ public class StopNodeParam implements Serializable {
      * 节点名称
      */
     private String nodeName;
+
+    private String runId;
 }

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/plugin/FlowExtendedConfiguration.java

@@ -22,7 +22,12 @@ public class FlowExtendedConfiguration {
     /**
      * 公共参数
      */
-    public String publicParam;
+    private String publicParam;
+
+    /**
+     * 是否支持多实例   0  支持  1 不支持
+     */
+    private String hasManyInstance;
 
     /**
      * 节点嵌入标记  a:b:c

byit-myth-core/myth-dto-core/src/main/java/com/byit/dto/specials/SpecialJavaNode.java

@@ -33,4 +33,6 @@ public class SpecialJavaNode implements Serializable {
     private String thisLogUrlPath;
 
     private String workspaceName;
+
+    private String runId;
 }