liteflow学习二

路行的亚洲

发布于 2023-10-08 13:37:21

73600

代码可运行

文章被收录于专栏：后端技术学习后端技术学习

运行总次数：0

代码可运行

前面我们看到了执行的具体逻辑，但是在哪里构建EL表达式的、获取的、执行的呢？

一、EL表达式的设置

在LiteFlowChainELBuilder可以看到EL表达式的逻辑是基于阿里开源的QLExpress实现的, 在初始化QLExpress的Runner中，作者扩展了很多操作Operator。在setEL(String elStr)这个方法中，我们可以会将FlowBus中的chainMap和NodeMap放入到Context中，也即DefaultContext，同时放入当前的chainId。接着会解析el表达式，解析成最外部一个condition, 内部可以嵌套很多层。

可以看到构建EL表达式的过程中会涉及到chainMap和NodeMap放入到DefaultContext中：

DefaultContext<String, Object> context = new DefaultContext<>();
// 这里一定要先放chain，再放node，因为node优先于chain，所以当重名时，node会覆盖掉chain
FlowBus.getChainMap().values().forEach(chain -> context.put(chain.getChainId(), chain));

// 往上下文里放入所有的node，使得el表达式可以直接引用到nodeId
FlowBus.getNodeMap().keySet().forEach(nodeId -> context.put(nodeId, FlowBus.getNode(nodeId)));

// 放入当前主chain的ID
context.put(ChainConstant.CURR_CHAIN_ID, this.chain.getChainId());

// 解析el成为一个Condition
// 为什么这里只是一个Condition，而不是一个List<Condition>呢
// 这里无论多复杂的，外面必定有一个最外层的Condition，所以这里只有一个，内部可以嵌套很多层，这点和以前的不太一样
Condition condition = (Condition) EXPRESS_RUNNER.execute(elStr, context, errorList, true, true);

// 把主要的condition加入
this.conditionList.add(condition);

可以看到我们的WHEN表达式会进行WHEN的EL表达式的解析，此时经过BaseOperator会对对应的Operator进行依次构建。所有的Operator都会经过com.yomahub.liteflow.builder.el.operator.base.BaseOperator#executeInner进行build。

因此我们来看看WhenOperator是如何构建的。

public WhenCondition build(Object[] objects) throws Exception {
        OperatorHelper.checkObjectSizeGtZero(objects);
        WhenCondition whenCondition = new WhenCondition();
        for (Object obj : objects) {
            whenCondition.addExecutable(OperatorHelper.convert(obj, Executable.class));
        }
        return whenCondition;
    }

此时会将可执行的执行器放入到executableGroup中，方便执行condition的时候获取执行。

public void addExecutable(String groupKey, Executable executable) {
        System.out.println("添加执行的operator");
        if (ObjectUtil.isNull(executable)) {
            return;
        }
        List<Executable> executableList = this.executableGroup.get(groupKey);
        if (CollUtil.isEmpty(executableList)) {
            this.executableGroup.put(groupKey, ListUtil.toList(executable));
        }
        else {
            this.executableGroup.get(groupKey).add(executable);
        }
    }

前面我们已经知道，liteflow的执行逻辑是先获取chain，然后执行chain，然后执行condition，最终执行业务逻辑中的nodeComponent。因此我们来看看并行器是如何实现的。

二、并行器是如何实现

前面说到THEN打头的是串行器，而WHEN打头的是并行器，因此我们可以看到Condition中的并行器对应的WhenCondition。

可以看到执行的execution是executeAsyncCondition(slotIndex)。

前面看到初始化LiteflowConfig的时候，没有配置信息。可以看到线程池的配置信息就是配置在LiteflowConfig中的。并行编排的实现借助了封装CompletableFuture对象。

前面我们知道执行condition的方法是executeCondition(Integer slotIndex)。执行condition的核心方法是com.yomahub.liteflow.flow.element.condition.WhenCondition#executeAsyncCondition。

那么此时我们似乎在Future的代码中找不到调用的逻辑，如果看过我上一篇文章的话，那应该知道调用condition之后，会调用具体的node执行业务逻辑编排代码。因此此时我们可以根据之前的线索知道会调用node中的execute方法。因此我们可以查询引用的地方，从而找到对应的地方。

根据前的ThenCondition的调用，我们知道调用的过程是condition到node的过程，因此这个过程必然会涉及到node的调用过程。因此我们可以知道对应的node的执行是在ParallelSupplier中。

public WhenFutureObj get() {
        try {
            executableItem.setCurrChainId(currChainId);
            executableItem.execute(slotIndex);
            return WhenFutureObj.success(executableItem.getId());
        }
        catch (Exception e) {
            return WhenFutureObj.fail(executableItem.getId(), e);
        }
    }

从而调用node执行具体的业务逻辑方法com.yomahub.liteflow.flow.element.Node#execute。

从而进一步调用我们自己写的业务逻辑。

总结下来即：组装数据chain、node ，构建EL表达式，然后根据通过槽拿到chain，从而进一步拿到condition，从而拿到node，调用业务系统继承NodeComponent重写的process方法。