!214 WHEN 增加 must 语法

Merge pull request !214 from luoyi/issues/I7XAIB
2026-05-14 20:22:07 +08:00 · 2023-09-21 15:00:06 +00:00
parent 4e0348ff39 d7c6383b09
commit 7b004bf1f5
16 changed files with 655 additions and 148 deletions
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/LiteFlowChainELBuilder.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/LiteFlowChainELBuilder.java
@@ -15,8 +15,8 @@ import com.yomahub.liteflow.exception.ELParseException;
 import com.yomahub.liteflow.exception.FlowSystemException;
 import com.yomahub.liteflow.flow.FlowBus;
 import com.yomahub.liteflow.flow.element.Chain;
-import com.yomahub.liteflow.flow.element.Node;
 import com.yomahub.liteflow.flow.element.Condition;
+import com.yomahub.liteflow.flow.element.Node;
 import com.yomahub.liteflow.log.LFLog;
 import com.yomahub.liteflow.log.LFLoggerManager;

@@ -72,6 +72,7 @@ public class LiteFlowChainELBuilder {
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.DEFAULT, Object.class, new DefaultOperator());
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.TAG, Object.class, new TagOperator());
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.ANY, Object.class, new AnyOperator());
+		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.MUST, Object.class, new MustOperator());
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.ID, Object.class, new IdOperator());
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.IGNORE_ERROR, Object.class, new IgnoreErrorOperator());
 		EXPRESS_RUNNER.addFunctionAndClassMethod(ChainConstant.THREAD_POOL, Object.class, new ThreadPoolOperator());
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/operator/AnyOperator.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/operator/AnyOperator.java
@@ -2,6 +2,7 @@ package com.yomahub.liteflow.builder.el.operator;

 import com.yomahub.liteflow.builder.el.operator.base.BaseOperator;
 import com.yomahub.liteflow.builder.el.operator.base.OperatorHelper;
+import com.yomahub.liteflow.enums.ParallelStrategyEnum;
 import com.yomahub.liteflow.flow.element.condition.WhenCondition;

 /**
@@ -19,7 +20,7 @@ public class AnyOperator extends BaseOperator<WhenCondition> {
 		WhenCondition whenCondition = OperatorHelper.convert(objects[0], WhenCondition.class);

 		Boolean any = OperatorHelper.convert(objects[1], Boolean.class);
-		whenCondition.setAny(any);
+		whenCondition.setParallelStrategy(any ? ParallelStrategyEnum.ANY : ParallelStrategyEnum.ALL);
 		return whenCondition;
 	}

--- a/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/operator/MustOperator.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/builder/el/operator/MustOperator.java
@@ -0,0 +1,28 @@
+package com.yomahub.liteflow.builder.el.operator;
+
+import com.yomahub.liteflow.builder.el.operator.base.BaseOperator;
+import com.yomahub.liteflow.builder.el.operator.base.OperatorHelper;
+import com.yomahub.liteflow.enums.ParallelStrategyEnum;
+import com.yomahub.liteflow.flow.element.condition.WhenCondition;
+
+/**
+ * EL 规则中的 must 的操作符
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class MustOperator extends BaseOperator<WhenCondition> {
+
+	@Override
+	public WhenCondition build(Object[] objects) throws Exception {
+		OperatorHelper.checkObjectSizeEqTwo(objects);
+
+		WhenCondition whenCondition = OperatorHelper.convert(objects[0], WhenCondition.class);
+
+		String specifyId = OperatorHelper.convert(objects[1], String.class);
+		whenCondition.setSpecifyId(specifyId);
+		whenCondition.setParallelStrategy(ParallelStrategyEnum.SPECIFY);
+		return whenCondition;
+	}
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/common/ChainConstant.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/common/ChainConstant.java
@@ -30,6 +30,8 @@ public interface ChainConstant {

 	String ANY = "any";

+	String MUST = "must";
+
 	String TYPE = "type";

 	String THEN = "THEN";
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/enums/ParallelStrategyEnum.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/enums/ParallelStrategyEnum.java
@@ -0,0 +1,57 @@
+package com.yomahub.liteflow.enums;
+
+import com.yomahub.liteflow.flow.parallel.strategy.AllOfParallelExecutor;
+import com.yomahub.liteflow.flow.parallel.strategy.AnyOfParallelExecutor;
+import com.yomahub.liteflow.flow.parallel.strategy.ParallelStrategyExecutor;
+import com.yomahub.liteflow.flow.parallel.strategy.SpecifyParallelExecutor;
+
+/**
+ * 并行策略枚举类
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public enum ParallelStrategyEnum {
+
+    ANY("anyOf", "完成任一任务", AnyOfParallelExecutor.class),
+
+    ALL("allOf", "完成全部任务", AllOfParallelExecutor.class),
+
+    SPECIFY("must", "完成指定 ID 任务", SpecifyParallelExecutor.class);
+
+    private String strategyType;
+
+    private String description;
+
+    private Class<? extends ParallelStrategyExecutor> clazz;
+
+    ParallelStrategyEnum(String strategyType, String description, Class<? extends ParallelStrategyExecutor> clazz) {
+        this.strategyType = strategyType;
+        this.description = description;
+        this.clazz = clazz;
+    }
+
+    public String getStrategyType() {
+        return strategyType;
+    }
+
+    public void setStrategyType(String strategyType) {
+        this.strategyType = strategyType;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    public void setDescription(String description) {
+        this.description = description;
+    }
+
+    public Class<? extends ParallelStrategyExecutor> getClazz() {
+        return clazz;
+    }
+
+    public void setClazz(Class<? extends ParallelStrategyExecutor> clazz) {
+        this.clazz = clazz;
+    }
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/exception/ParallelExecutorCreateException.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/exception/ParallelExecutorCreateException.java
@@ -0,0 +1,29 @@
+package com.yomahub.liteflow.exception;
+
+/**
+ * 并行策略执行器创建异常
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class ParallelExecutorCreateException extends RuntimeException {
+
+	private static final long serialVersionUID = 1L;
+
+	/** 异常信息 */
+	private String message;
+
+	public ParallelExecutorCreateException(String message) {
+		this.message = message;
+	}
+
+	@Override
+	public String getMessage() {
+		return message;
+	}
+
+	public void setMessage(String message) {
+		this.message = message;
+	}
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/element/condition/WhenCondition.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/element/condition/WhenCondition.java
@@ -7,28 +7,17 @@
 */
 package com.yomahub.liteflow.flow.element.condition;

-import cn.hutool.core.util.ObjectUtil;
-import cn.hutool.core.util.StrUtil;
 import com.yomahub.liteflow.common.LocalDefaultFlowConstant;
 import com.yomahub.liteflow.enums.ConditionTypeEnum;
-import com.yomahub.liteflow.exception.WhenExecuteException;
+import com.yomahub.liteflow.enums.ParallelStrategyEnum;
 import com.yomahub.liteflow.flow.element.Condition;
-import com.yomahub.liteflow.flow.parallel.CompletableFutureTimeout;
-import com.yomahub.liteflow.flow.parallel.ParallelSupplier;
-import com.yomahub.liteflow.flow.parallel.WhenFutureObj;
+import com.yomahub.liteflow.flow.parallel.strategy.ParallelStrategyExecutor;
+import com.yomahub.liteflow.flow.parallel.strategy.ParallelStrategyHelper;
 import com.yomahub.liteflow.log.LFLog;
 import com.yomahub.liteflow.log.LFLoggerManager;
-import com.yomahub.liteflow.property.LiteflowConfig;
-import com.yomahub.liteflow.property.LiteflowConfigGetter;
-import com.yomahub.liteflow.slot.DataBus;
-import com.yomahub.liteflow.slot.Slot;
 import com.yomahub.liteflow.thread.ExecutorHelper;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
+
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;

 /**
 * 并行器
@@ -46,8 +35,11 @@ public class WhenCondition extends Condition {
 	// 此属性已弃用
 	private String group = LocalDefaultFlowConstant.DEFAULT;

-	// 只在when类型下有效，为true的话说明在多个并行节点下，任意一个成功，整个when就成功
-	private boolean any = false;
+	// 当前 When 对应并行策略，默认为 ALL
+	private ParallelStrategyEnum parallelStrategy;
+
+	// 只有 must 条件下，才会赋值 specifyId
+	private String specifyId;

 	// when单独的线程池名称
 	private String threadExecutorClass;
@@ -68,136 +60,16 @@ public class WhenCondition extends Condition {
 		return ConditionTypeEnum.TYPE_WHEN;
 	}

-	// 使用线程池执行when并发流程
+	// 使用线程池执行 when 并发流程
 	// 这块涉及到挺多的多线程逻辑，所以注释比较详细，看到这里的童鞋可以仔细阅读
 	private void executeAsyncCondition(Integer slotIndex) throws Exception {
-		Slot slot = DataBus.getSlot(slotIndex);

-		String currChainName = this.getCurrChainId();
+		// 获取并发执行策略
+		ParallelStrategyExecutor parallelStrategyExecutor = ParallelStrategyHelper.loadInstance().buildParallelExecutor(this.getParallelStrategy());

-		// 此方法其实只会初始化一次Executor，不会每次都会初始化。Executor是唯一的
-		ExecutorService parallelExecutor = ExecutorHelper.loadInstance()
-				.buildWhenExecutor(this.getThreadExecutorClass());
+		// 执行并发逻辑
+		parallelStrategyExecutor.execute(this, slotIndex);

-		// 获得liteflow的参数
-		LiteflowConfig liteflowConfig = LiteflowConfigGetter.get();
-
-		// 定义是否中断参数
-		// 这里为什么要定义成数组呢，因为后面lambda要用到，根据final不能修改引用的原则，这里用了数组对象
-		final boolean[] interrupted = { false };
-
-		// 这里主要是做了封装CompletableFuture对象，用lumbda表达式做了很多事情，这句代码要仔细理清
-		// 1.先进行过滤，前置和后置组件过滤掉，因为在EL Chain处理的时候已经提出来了
-		// 2.过滤isAccess为false的情况，因为不过滤这个的话，如果加上了any，那么isAccess为false那就是最快的了
-		// 3.根据condition.getNodeList()的集合进行流处理，用map进行把executable对象转换成List<CompletableFuture<WhenFutureObj>>
-		// 4.在转的过程中，套入CompletableFutureTimeout方法进行超时判断，如果超时则用WhenFutureObj.timeOut返回超时的对象
-		// 5.第2个参数是主要的本体CompletableFuture，传入了ParallelSupplier和线程池对象
-		if (ObjectUtil.isNull(this.getMaxWaitTime())) {
-			if (ObjectUtil.isNotNull(liteflowConfig.getWhenMaxWaitSeconds())) {
-				// 获取全局异步线程最长等待秒数
-				this.setMaxWaitTime(liteflowConfig.getWhenMaxWaitSeconds());
-				this.setMaxWaitTimeUnit(TimeUnit.SECONDS);
-			} else {
-				// 获取全局异步线程最⻓的等待时间
-				this.setMaxWaitTime(liteflowConfig.getWhenMaxWaitTime());
-			}
-		}
-
-		if (ObjectUtil.isNull(this.getMaxWaitTimeUnit())) {
-			// 获取全局异步线程最⻓的等待时间单位
-			this.setMaxWaitTimeUnit(liteflowConfig.getWhenMaxWaitTimeUnit());
-		}
-
-		List<CompletableFuture<WhenFutureObj>> completableFutureList = this.getExecutableList()
-				.stream()
-				.filter(executable -> !(executable instanceof PreCondition) && !(executable instanceof FinallyCondition))
-				.filter(executable -> {
-					try {
-						return executable.isAccess(slotIndex);
-					} catch (Exception e) {
-						LOG.error("there was an error when executing the when component isAccess", e);
-						return false;
-					}
-				})
-				.map(executable -> CompletableFutureTimeout.completeOnTimeout(
-						WhenFutureObj.timeOut(executable.getId()),
-						CompletableFuture.supplyAsync(new ParallelSupplier(executable, currChainName, slotIndex),
-								parallelExecutor),
-						this.getMaxWaitTime(), this.getMaxWaitTimeUnit()))
-				.collect(Collectors.toList());
-
-		CompletableFuture<?> resultCompletableFuture;
-
-		// 这里判断执行方式
-		// 如果any为false，说明这些异步任务全部执行好或者超时，才返回
-		// 如果any为true，说明这些异步任务只要任意一个执行完成，就返回
-		if (this.isAny()) {
-			// 把这些CompletableFuture通过anyOf合成一个CompletableFuture
-			resultCompletableFuture = CompletableFuture
-					.anyOf(completableFutureList.toArray(new CompletableFuture[] {}));
-		} else {
-			// 把这些CompletableFuture通过allOf合成一个CompletableFuture
-			resultCompletableFuture = CompletableFuture
-					.allOf(completableFutureList.toArray(new CompletableFuture[] {}));
-		}
-
-		try {
-			// 进行执行，这句执行完后，就意味着所有的任务要么执行完毕，要么超时返回
-			resultCompletableFuture.get();
-		} catch (InterruptedException | ExecutionException e) {
-			LOG.error("there was an error when executing the CompletableFuture", e);
-			interrupted[0] = true;
-		}
-
-		// 拿到已经完成的CompletableFuture
-		// 如果any为false，那么所有任务都已经完成
-		// 如果any为true，那么这里拿到的是第一个完成的任务
-		// 这里过滤和转换一起用lumbda做了
-		List<WhenFutureObj> allCompletableWhenFutureObjList = completableFutureList.stream().filter(f -> {
-			// 过滤出已经完成的，没完成的就直接终止
-			if (f.isDone()) {
-				return true;
-			} else {
-				f.cancel(true);
-				return false;
-			}
-		}).map(f -> {
-			try {
-				return f.get();
-			} catch (InterruptedException | ExecutionException e) {
-				interrupted[0] = true;
-				return null;
-			}
-		}).collect(Collectors.toList());
-
-		// 判断超时，上面已经拿到了所有已经完成的CompletableFuture
-		// 那我们只要过滤出超时的CompletableFuture
-		List<WhenFutureObj> timeOutWhenFutureObjList = allCompletableWhenFutureObjList.stream()
-				.filter(WhenFutureObj::isTimeout)
-				.collect(Collectors.toList());
-
-		// 输出超时信息
-		timeOutWhenFutureObjList.forEach(whenFutureObj -> LOG.warn(
-				"executing thread has reached max-wait-seconds, thread canceled.Execute-item: [{}]", whenFutureObj.getExecutorName()));
-
-		// 当配置了ignoreError = false，出现interrupted或者!f.get()的情况，将抛出WhenExecuteException
-		if (!this.isIgnoreError()) {
-			if (interrupted[0]) {
-				throw new WhenExecuteException(StrUtil
-						.format("requestId [{}] when execute interrupted. errorResume [false].", slot.getRequestId()));
-			}
-
-			// 循环判断CompletableFuture的返回值，如果异步执行失败，则抛出相应的业务异常
-			for (WhenFutureObj whenFutureObj : allCompletableWhenFutureObjList) {
-				if (!whenFutureObj.isSuccess()) {
-					LOG.info(StrUtil.format("when-executor[{}] execute failed. errorResume [false].", whenFutureObj.getExecutorName()));
-					throw whenFutureObj.getEx();
-				}
-			}
-		} else if (interrupted[0]) {
-			// 这里由于配置了ignoreError，所以只打印warn日志
-			LOG.warn("executing when condition timeout , but ignore with errorResume.");
-		}
 	}

 	public boolean isIgnoreError() {
@@ -216,12 +88,20 @@ public class WhenCondition extends Condition {
 		this.group = group;
 	}

-	public boolean isAny() {
-		return any;
+	public ParallelStrategyEnum getParallelStrategy() {
+		return parallelStrategy;
 	}

-	public void setAny(boolean any) {
-		this.any = any;
+	public void setParallelStrategy(ParallelStrategyEnum parallelStrategy) {
+		this.parallelStrategy = parallelStrategy;
+	}
+
+	public String getSpecifyId() {
+		return specifyId;
+	}
+
+	public void setSpecifyId(String specifyId) {
+		this.specifyId = specifyId;
 	}

 	public String getThreadExecutorClass() {
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/AllOfParallelExecutor.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/AllOfParallelExecutor.java
@@ -0,0 +1,31 @@
+package com.yomahub.liteflow.flow.parallel.strategy;
+
+import com.yomahub.liteflow.flow.element.condition.WhenCondition;
+import com.yomahub.liteflow.flow.parallel.WhenFutureObj;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * 完成全部任务
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class AllOfParallelExecutor extends ParallelStrategyExecutor {
+
+    @Override
+    public void execute(WhenCondition whenCondition, Integer slotIndex) throws Exception {
+
+        // 获取所有 CompletableFuture 任务
+        List<CompletableFuture<WhenFutureObj>> whenAllTaskList = this.getWhenAllTaskList(whenCondition, slotIndex);
+
+        // 把这些 CompletableFuture 通过 allOf 合成一个 CompletableFuture，表明完成所有任务
+        CompletableFuture<?> specifyTask = CompletableFuture.allOf(whenAllTaskList.toArray(new CompletableFuture[] {}));
+
+        // 结果处理
+        this.handleTaskResult(whenCondition, slotIndex, whenAllTaskList, specifyTask);
+
+    }
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/AnyOfParallelExecutor.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/AnyOfParallelExecutor.java
@@ -0,0 +1,31 @@
+package com.yomahub.liteflow.flow.parallel.strategy;
+
+import com.yomahub.liteflow.flow.element.condition.WhenCondition;
+import com.yomahub.liteflow.flow.parallel.WhenFutureObj;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * 完成任一任务
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class AnyOfParallelExecutor extends ParallelStrategyExecutor {
+
+    @Override
+    public void execute(WhenCondition whenCondition, Integer slotIndex) throws Exception {
+
+        // 获取所有 CompletableFuture 任务
+        List<CompletableFuture<WhenFutureObj>> whenAllTaskList = this.getWhenAllTaskList(whenCondition, slotIndex);
+
+        // 把这些 CompletableFuture 通过 anyOf 合成一个 CompletableFuture，表明完成任一任务
+        CompletableFuture<?> specifyTask = CompletableFuture.anyOf(whenAllTaskList.toArray(new CompletableFuture[] {}));
+
+        // 结果处理
+        this.handleTaskResult(whenCondition, slotIndex, whenAllTaskList, specifyTask);
+
+    }
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/ParallelStrategyExecutor.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/ParallelStrategyExecutor.java
@@ -0,0 +1,199 @@
+package com.yomahub.liteflow.flow.parallel.strategy;
+
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.StrUtil;
+import com.yomahub.liteflow.enums.ParallelStrategyEnum;
+import com.yomahub.liteflow.exception.WhenExecuteException;
+import com.yomahub.liteflow.flow.element.Executable;
+import com.yomahub.liteflow.flow.element.condition.FinallyCondition;
+import com.yomahub.liteflow.flow.element.condition.PreCondition;
+import com.yomahub.liteflow.flow.element.condition.WhenCondition;
+import com.yomahub.liteflow.flow.parallel.CompletableFutureTimeout;
+import com.yomahub.liteflow.flow.parallel.ParallelSupplier;
+import com.yomahub.liteflow.flow.parallel.WhenFutureObj;
+import com.yomahub.liteflow.log.LFLog;
+import com.yomahub.liteflow.log.LFLoggerManager;
+import com.yomahub.liteflow.property.LiteflowConfig;
+import com.yomahub.liteflow.property.LiteflowConfigGetter;
+import com.yomahub.liteflow.slot.DataBus;
+import com.yomahub.liteflow.slot.Slot;
+import com.yomahub.liteflow.thread.ExecutorHelper;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * 并发策略执行器抽象类
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public abstract class ParallelStrategyExecutor {
+
+    protected final LFLog LOG = LFLoggerManager.getLogger(this.getClass());
+
+    /**
+     * 封装 CompletableFuture 对象
+     * @param executable
+     * @param parallelExecutor
+     * @param whenCondition
+     * @param currChainName
+     * @param slotIndex
+     * @return
+     */
+    protected CompletableFuture<WhenFutureObj> wrappedFutureObj(Executable executable, ExecutorService parallelExecutor,
+                                                                WhenCondition whenCondition, String currChainName, Integer slotIndex) {
+        // 套入 CompletableFutureTimeout 方法进行超时判断，如果超时则用 WhenFutureObj.timeOut 返回超时的对象
+        // 第 2 个参数是主要的本体 CompletableFuture，传入了 ParallelSupplier 和线程池对象
+        return CompletableFutureTimeout.completeOnTimeout(
+                WhenFutureObj.timeOut(executable.getId()),
+                CompletableFuture.supplyAsync(new ParallelSupplier(executable, currChainName, slotIndex), parallelExecutor),
+                whenCondition.getMaxWaitTime(),
+                whenCondition.getMaxWaitTimeUnit());
+    }
+
+    /**
+     * 设置 WhenCondition 参数
+     * @param whenCondition
+     */
+    protected void setWhenConditionParams(WhenCondition whenCondition) {
+        // 获得 liteflow 的参数
+        LiteflowConfig liteflowConfig = LiteflowConfigGetter.get();
+        if (ObjectUtil.isNull(whenCondition.getMaxWaitTime())) {
+            if (ObjectUtil.isNotNull(liteflowConfig.getWhenMaxWaitSeconds())) {
+                // 获取全局异步线程最长等待秒数
+                whenCondition.setMaxWaitTime(liteflowConfig.getWhenMaxWaitSeconds());
+                whenCondition.setMaxWaitTimeUnit(TimeUnit.SECONDS);
+            } else {
+                // 获取全局异步线程最⻓的等待时间
+                whenCondition.setMaxWaitTime(liteflowConfig.getWhenMaxWaitTime());
+            }
+        }
+
+        if (ObjectUtil.isNull(whenCondition.getMaxWaitTimeUnit())) {
+            // 获取全局异步线程最⻓的等待时间单位
+            whenCondition.setMaxWaitTimeUnit(liteflowConfig.getWhenMaxWaitTimeUnit());
+        }
+    }
+
+    /**
+     * 获取所有任务 CompletableFuture 集合
+     * @param whenCondition
+     * @param slotIndex
+     * @return
+     */
+    protected List<CompletableFuture<WhenFutureObj>> getWhenAllTaskList(WhenCondition whenCondition, Integer slotIndex) {
+
+        String currChainName = whenCondition.getCurrChainId();
+
+        // 此方法其实只会初始化一次 Executor，不会每次都会初始化。Executor是唯一的
+        ExecutorService parallelExecutor = ExecutorHelper.loadInstance().buildWhenExecutor(whenCondition.getThreadExecutorClass());
+
+        // 设置 whenCondition 参数
+        setWhenConditionParams(whenCondition);
+
+        // 这里主要是做了封装 CompletableFuture 对象，用 lumbda 表达式做了很多事情，这句代码要仔细理清
+        // 1.先进行过滤，前置和后置组件过滤掉，因为在 EL Chain 处理的时候已经提出来了
+        // 2.过滤 isAccess 为 false 的情况，因为不过滤这个的话，如果加上了 any，那么 isAccess 为 false 那就是最快的了
+        // 3.根据 condition.getNodeList() 的集合进行流处理，用 map 进行把 executable 对象转换成 List<CompletableFuture<WhenFutureObj>>
+        List<CompletableFuture<WhenFutureObj>> completableFutureList = whenCondition.getExecutableList()
+                .stream()
+                .filter(executable -> !(executable instanceof PreCondition) && !(executable instanceof FinallyCondition))
+                .filter(executable -> {
+                    try {
+                        return executable.isAccess(slotIndex);
+                    } catch (Exception e) {
+                        LOG.error("there was an error when executing the when component isAccess", e);
+                        return false;
+                    }
+                })
+                .map(executable -> wrappedFutureObj(executable, parallelExecutor, whenCondition, currChainName, slotIndex))
+                .collect(Collectors.toList());
+
+        return completableFutureList;
+    }
+
+    /**
+     * 任务结果处理
+     * @param whenCondition 并行组件对象
+     * @param slotIndex 当前 slot 的 index
+     * @param whenAllTaskList 并行组件中所有任务列表
+     * @param specifyTask 指定预先完成的任务，详见 {@link ParallelStrategyEnum}
+     * @throws Exception
+     */
+    protected void handleTaskResult(WhenCondition whenCondition, Integer slotIndex, List<CompletableFuture<WhenFutureObj>> whenAllTaskList,
+                                    CompletableFuture<?> specifyTask) throws Exception {
+
+        Slot slot = DataBus.getSlot(slotIndex);
+
+        // 定义是否中断参数
+        // 这里为什么要定义成数组呢，因为后面 lambda 要用到，根据 final 不能修改引用的原则，这里用了数组对象
+        final boolean[] interrupted = { false };
+
+        try {
+            // 进行执行，这句执行完后有三种可能，所有任务执行完成、任一任务执行完成、指定的任务执行完成
+            specifyTask.get();
+        } catch (InterruptedException | ExecutionException e) {
+            LOG.error("there was an error when executing the CompletableFuture", e);
+            interrupted[0] = true;
+        }
+
+        // 拿到已经完成的 CompletableFuture 对象
+        // 如果 any 为 false，那么所有任务都已经完成
+        // 如果 any 为 true，那么这里拿到的是第一个完成的任务
+        // 如果为 must，那么这里获取到的就是指定的任务
+        // 这里过滤和转换一起用 lambda 做了
+        List<WhenFutureObj> allCompletableWhenFutureObjList = whenAllTaskList.stream().filter(f -> {
+            // 过滤出已经完成的，没完成的就直接终止
+            if (f.isDone()) {
+                return true;
+            } else {
+                f.cancel(true);
+                return false;
+            }
+        }).map(f -> {
+            try {
+                return f.get();
+            } catch (InterruptedException | ExecutionException e) {
+                interrupted[0] = true;
+                return null;
+            }
+        }).collect(Collectors.toList());
+
+        // 判断超时，上面已经拿到了所有已经完成的 CompletableFuture
+        // 那我们只要过滤出超时的 CompletableFuture
+        List<WhenFutureObj> timeOutWhenFutureObjList = allCompletableWhenFutureObjList.stream()
+                .filter(WhenFutureObj::isTimeout)
+                .collect(Collectors.toList());
+
+        // 输出超时信息
+        timeOutWhenFutureObjList.forEach(whenFutureObj -> LOG.warn(
+                "executing thread has reached max-wait-seconds, thread canceled.Execute-item: [{}]", whenFutureObj.getExecutorName()));
+
+        // 当配置了 ignoreError = false，出现 interrupted 或者 !f.get() 的情况，将抛出 WhenExecuteException
+        if (!whenCondition.isIgnoreError()) {
+            if (interrupted[0]) {
+                throw new WhenExecuteException(StrUtil
+                        .format("requestId [{}] when execute interrupted. errorResume [false].", slot.getRequestId()));
+            }
+
+            // 循环判断CompletableFuture的返回值，如果异步执行失败，则抛出相应的业务异常
+            for (WhenFutureObj whenFutureObj : allCompletableWhenFutureObjList) {
+                if (!whenFutureObj.isSuccess()) {
+                    LOG.info(StrUtil.format("when-executor[{}] execute failed. errorResume [false].", whenFutureObj.getExecutorName()));
+                    throw whenFutureObj.getEx();
+                }
+            }
+        } else if (interrupted[0]) {
+            // 这里由于配置了 ignoreError，所以只打印 warn 日志
+            LOG.warn("executing when condition timeout , but ignore with errorResume.");
+        }
+    }
+
+    public abstract void execute(WhenCondition whenCondition, Integer slotIndex) throws Exception;
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/ParallelStrategyHelper.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/ParallelStrategyHelper.java
@@ -0,0 +1,79 @@
+package com.yomahub.liteflow.flow.parallel.strategy;
+
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.util.ObjUtil;
+import com.yomahub.liteflow.enums.ParallelStrategyEnum;
+import com.yomahub.liteflow.exception.ParallelExecutorCreateException;
+import com.yomahub.liteflow.log.LFLog;
+import com.yomahub.liteflow.log.LFLoggerManager;
+import com.yomahub.liteflow.spi.holder.ContextAwareHolder;
+
+import java.util.Map;
+
+/**
+ * WHEN 并发策略辅助
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class ParallelStrategyHelper {
+
+    private final LFLog LOG = LFLoggerManager.getLogger(ParallelStrategyHelper.class);
+
+    /**
+     * 此处使用Map缓存线程池信息 key - 线程池构建者的Class全类名 value - 线程池对象
+     */
+    private final Map<ParallelStrategyEnum, ParallelStrategyExecutor> strategyExecutorMap;
+
+    private ParallelStrategyHelper() {
+        strategyExecutorMap = MapUtil.newConcurrentHashMap();
+    }
+
+    /**
+     * 使用静态内部类实现单例模式
+     */
+    private static class Holder {
+
+        static final ParallelStrategyHelper INSTANCE = new ParallelStrategyHelper();
+
+    }
+
+    public static ParallelStrategyHelper loadInstance() {
+        return ParallelStrategyHelper.Holder.INSTANCE;
+    }
+
+    private ParallelStrategyExecutor getParallelStrategyExecutor(ParallelStrategyEnum parallelStrategyEnum) {
+        try {
+            ParallelStrategyExecutor strategyExecutor = strategyExecutorMap.get(parallelStrategyEnum);
+            if (ObjUtil.isNotNull(strategyExecutor)) return strategyExecutor;
+
+            Class<ParallelStrategyExecutor> executorClass = (Class<ParallelStrategyExecutor>) Class.forName(parallelStrategyEnum.getClazz().getName());
+            strategyExecutor = ContextAwareHolder.loadContextAware().registerBean(executorClass);
+            strategyExecutorMap.put(parallelStrategyEnum, strategyExecutor);
+            return strategyExecutor;
+        } catch (Exception e) {
+            LOG.error(e.getMessage());
+            throw new ParallelExecutorCreateException(e.getMessage());
+        }
+    }
+
+    public ParallelStrategyExecutor buildParallelExecutor(ParallelStrategyEnum parallelStrategyEnum) {
+        if (ObjUtil.isNull(parallelStrategyEnum)) return buildParallelExecutor();
+        return getParallelStrategyExecutor(parallelStrategyEnum);
+    }
+
+    /**
+     * 默认需完成所有任务
+     * @return
+     */
+    public ParallelStrategyExecutor buildParallelExecutor() {
+        return buildParallelExecutor(ParallelStrategyEnum.ALL);
+    }
+
+    public void clearStrategyExecutorMap() {
+        if (MapUtil.isNotEmpty(strategyExecutorMap)) {
+            strategyExecutorMap.clear();
+        }
+    }
+
+}
--- a/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/SpecifyParallelExecutor.java
+++ b/liteflow-core/src/main/java/com/yomahub/liteflow/flow/parallel/strategy/SpecifyParallelExecutor.java
@@ -0,0 +1,75 @@
+package com.yomahub.liteflow.flow.parallel.strategy;
+
+import cn.hutool.core.util.ObjUtil;
+import com.yomahub.liteflow.flow.element.condition.FinallyCondition;
+import com.yomahub.liteflow.flow.element.condition.PreCondition;
+import com.yomahub.liteflow.flow.element.condition.WhenCondition;
+import com.yomahub.liteflow.flow.parallel.WhenFutureObj;
+import com.yomahub.liteflow.thread.ExecutorHelper;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * 完成指定任务执行器，使用 ID 进行比较
+ *
+ * @author luo yi
+ * @since 2.11.0
+ */
+public class SpecifyParallelExecutor extends ParallelStrategyExecutor {
+
+    @Override
+    public void execute(WhenCondition whenCondition, Integer slotIndex) throws Exception {
+
+        String currChainName = whenCondition.getCurrChainId();
+
+        // 设置 whenCondition 参数
+        this.setWhenConditionParams(whenCondition);
+
+        // 此方法其实只会初始化一次Executor，不会每次都会初始化。Executor 是唯一的
+        ExecutorService parallelExecutor = ExecutorHelper.loadInstance().buildWhenExecutor(whenCondition.getThreadExecutorClass());
+
+        // 指定任务
+        final CompletableFuture<?>[] specifyTask = { null };
+
+        // 所有任务集合
+        List<CompletableFuture<WhenFutureObj>> allTaskList = new ArrayList<>();
+
+        // 遍历 when 所有 node，进行筛选及处理
+        whenCondition.getExecutableList()
+                .stream()
+                .filter(executable -> !(executable instanceof PreCondition) && !(executable instanceof FinallyCondition))
+                .filter(executable -> {
+                    try {
+                        return executable.isAccess(slotIndex);
+                    } catch (Exception e) {
+                        LOG.error("there was an error when executing the when component isAccess", e);
+                        return false;
+                    }
+                })
+                .forEach(executable -> {
+                    // 处理 task，封装成 CompletableFuture 对象
+                    CompletableFuture<WhenFutureObj> completableFutureTask = wrappedFutureObj(executable, parallelExecutor, whenCondition, currChainName, slotIndex);
+                    // 存在 must 指定 ID 的 task，且该任务只会有一个或者没有
+                    if (whenCondition.getSpecifyId().equals(executable.getId())) {
+                        // 设置指定任务 future 对象
+                        specifyTask[0] = completableFutureTask;
+                    }
+                    // 组合所有任务
+                    allTaskList.add(completableFutureTask);
+                });
+
+        if (ObjUtil.isNull(specifyTask[0])) {
+            LOG.warn("The specified task[{}] was not found, waiting for all tasks to complete by default.", whenCondition.getSpecifyId());
+            // 不存在指定任务，则需要等待所有任务都执行完成
+            specifyTask[0] = CompletableFuture.allOf(allTaskList.toArray(new CompletableFuture[] {}));
+        }
+
+        // 结果处理
+        this.handleTaskResult(whenCondition, slotIndex, allTaskList, specifyTask[0]);
+
+    }
+
+}
--- a/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/AsyncNodeELSpringbootTest.java
+++ b/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/AsyncNodeELSpringbootTest.java
@@ -139,4 +139,40 @@ public class AsyncNodeELSpringbootTest extends BaseTest {
 		Assertions.assertTrue(context.getData("check").toString().startsWith("habc"));
 	}

+	// 测试 must 关键字
+	@Test
+	public void testAsyncFlow9() throws Exception {
+		LiteflowResponse response = flowExecutor.execute2Resp("chain9", "it's a base request");
+		DefaultContext context = response.getFirstContextBean();
+		Assertions.assertTrue(response.isSuccess());
+		Assertions.assertTrue(context.getData("check").toString().startsWith("habc"));
+	}
+
+	// 测试 must 与 ignoreError 关键字，不忽略异常
+	@Test
+	public void testAsyncFlow10() throws Exception {
+		LiteflowResponse response = flowExecutor.execute2Resp("chain10", "it's a base request");
+		DefaultContext context = response.getFirstContextBean();
+		Assertions.assertTrue(context.getData("check").toString().startsWith("kg"));
+		Assertions.assertFalse(response.isSuccess());
+	}
+
+	// 测试 must 与 ignoreError 关键字，忽略异常
+	@Test
+	public void testAsyncFlow11() throws Exception {
+		LiteflowResponse response = flowExecutor.execute2Resp("chain11", "it's a base request");
+		DefaultContext context = response.getFirstContextBean();
+		Assertions.assertTrue(response.isSuccess());
+		Assertions.assertTrue(context.getData("check").toString().startsWith("kgdabc"));
+	}
+
+	// 测试 must 、 ignoreError 、 id 关键字
+	@Test
+	public void testAsyncFlow12() throws Exception {
+		LiteflowResponse response = flowExecutor.execute2Resp("chain12", "it's a base request");
+		DefaultContext context = response.getFirstContextBean();
+		Assertions.assertTrue(response.isSuccess());
+		Assertions.assertTrue(context.getData("check").toString().startsWith("akbc"));
+	}
+
 }
--- a/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/cmp/KCmp.java
+++ b/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/cmp/KCmp.java
@@ -0,0 +1,27 @@
+package com.yomahub.liteflow.test.asyncNode.cmp;
+
+import com.yomahub.liteflow.core.NodeComponent;
+import com.yomahub.liteflow.slot.DefaultContext;
+import org.springframework.stereotype.Component;
+
+@Component("k")
+public class KCmp extends NodeComponent {
+
+	@Override
+	public void process() throws Exception {
+		Thread.sleep(200);
+		DefaultContext context = this.getFirstContextBean();
+		synchronized (NodeComponent.class) {
+			if (context.hasData("check")) {
+				String str = context.getData("check");
+				str += this.getNodeId();
+				context.setData("check", str);
+			}
+			else {
+				context.setData("check", this.getNodeId());
+			}
+		}
+		System.out.println("Kcomp executed!");
+	}
+
+}
--- a/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/cmp/LCmp.java
+++ b/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/java/com/yomahub/liteflow/test/asyncNode/cmp/LCmp.java
@@ -0,0 +1,15 @@
+package com.yomahub.liteflow.test.asyncNode.cmp;
+
+import com.yomahub.liteflow.core.NodeComponent;
+import org.springframework.stereotype.Component;
+
+@Component("l")
+public class LCmp extends NodeComponent {
+
+	@Override
+	public void process() throws Exception {
+		System.out.println("Lcomp executed! Throw exception");
+		int i = 1/0;
+	}
+
+}
--- a/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/resources/asyncNode/flow.el.xml
+++ b/liteflow-testcase-el/liteflow-testcase-el-springboot/src/test/resources/asyncNode/flow.el.xml
@@ -46,4 +46,20 @@
        THEN(WHEN(d, g, h).any(true), THEN(a, b, c));
    </chain>

+    <chain name="chain9">
+        THEN(WHEN(d, g, h).must("h"), THEN(a, b, c));
+    </chain>
+
+    <chain name="chain10">
+        THEN(WHEN(d, g, k, l).must("g").ignoreError(false), THEN(a, b, c));
+    </chain>
+
+    <chain name="chain11">
+        THEN(WHEN(d, g, k, l).ignoreError(true).must("d"), THEN(a, b, c));
+    </chain>
+
+    <chain name="chain12">
+        THEN(WHEN(d, g, l, a, THEN(k, b).id("z")).ignoreError(true).must("z"), c);
+    </chain>
+
 </flow>