diff --git 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 index 97e3de3e5..a5e26cab9 100644 --- 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()); diff --git 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 index 144f3a275..4f326dd68 100644 --- 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 = 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; } diff --git 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 new file mode 100644 index 000000000..cb018c529 --- /dev/null +++ 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 { + + @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; + } + +} diff --git a/liteflow-core/src/main/java/com/yomahub/liteflow/common/ChainConstant.java b/liteflow-core/src/main/java/com/yomahub/liteflow/common/ChainConstant.java index 7379f26ea..a097cdfe1 100644 --- 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"; diff --git a/liteflow-core/src/main/java/com/yomahub/liteflow/enums/ParallelStrategyEnum.java b/liteflow-core/src/main/java/com/yomahub/liteflow/enums/ParallelStrategyEnum.java new file mode 100644 index 000000000..09c9af437 --- /dev/null +++ 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 clazz; + + ParallelStrategyEnum(String strategyType, String description, Class 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 getClazz() { + return clazz; + } + + public void setClazz(Class clazz) { + this.clazz = clazz; + } +} diff --git a/liteflow-core/src/main/java/com/yomahub/liteflow/exception/ParallelExecutorCreateException.java b/liteflow-core/src/main/java/com/yomahub/liteflow/exception/ParallelExecutorCreateException.java new file mode 100644 index 000000000..68ea8a39a --- /dev/null +++ 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; + } + +} diff --git 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 index 3555d37af..7422a495e 100644 --- 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> - // 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> 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 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 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() { diff --git 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 new file mode 100644 index 000000000..313e02d4d --- /dev/null +++ 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> whenAllTaskList = this.getWhenAllTaskList(whenCondition, slotIndex); + + // 把这些 CompletableFuture 通过 allOf 合成一个 CompletableFuture，表明完成所有任务 + CompletableFuture specifyTask = CompletableFuture.allOf(whenAllTaskList.toArray(new CompletableFuture[] {})); + + // 结果处理 + this.handleTaskResult(whenCondition, slotIndex, whenAllTaskList, specifyTask); + + } + +} diff --git 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 new file mode 100644 index 000000000..02fcc4646 --- /dev/null +++ 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> whenAllTaskList = this.getWhenAllTaskList(whenCondition, slotIndex); + + // 把这些 CompletableFuture 通过 anyOf 合成一个 CompletableFuture，表明完成任一任务 + CompletableFuture specifyTask = CompletableFuture.anyOf(whenAllTaskList.toArray(new CompletableFuture[] {})); + + // 结果处理 + this.handleTaskResult(whenCondition, slotIndex, whenAllTaskList, specifyTask); + + } + +} diff --git 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 new file mode 100644 index 000000000..95adb04ee --- /dev/null +++ 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 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> 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> + List> 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> 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 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 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; + +} diff --git 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 new file mode 100644 index 000000000..250fccfb5 --- /dev/null +++ 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 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 executorClass = (Class) 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(); + } + } + +} diff --git 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 new file mode 100644 index 000000000..9474232d4 --- /dev/null +++ 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> 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 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]); + + } + +} diff --git 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 index 30703e603..99f9e1d97 100644 --- 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")); + } + } diff --git 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 new file mode 100644 index 000000000..f4b27d714 --- /dev/null +++ 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!"); + } + +} diff --git 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 new file mode 100644 index 000000000..87e08565c --- /dev/null +++ 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; + } + +} diff --git 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 index 06373bb2f..e48bf988e 100644 --- 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)); + + THEN(WHEN(d, g, h).must("h"), THEN(a, b, c)); + + + + THEN(WHEN(d, g, k, l).must("g").ignoreError(false), THEN(a, b, c)); + + + + THEN(WHEN(d, g, k, l).ignoreError(true).must("d"), THEN(a, b, c)); + + + + THEN(WHEN(d, g, l, a, THEN(k, b).id("z")).ignoreError(true).must("z"), c); + + \ No newline at end of file