您现在的位置是：首页 > 文章详情

数据结构与算法的实际应用（有向图）——有依赖的任务并行处理框架

日期：2022-01-15点击：1041收藏

背景

一开始接到一个技改需求，需要将我们系统的一些首页查询接口进行优化，合并为一个聚合接口提供给前端；最初与前端同学进行交流，认为这些首页查询接口没有依赖关系，因此对于我来说解决方案比较简单，直接将需要查询的接口进行并行处理即可：

其中每一个查询抽象成了一个Function，在子线程中调用apply方法执行。

但是在与前端同学联调时，发现其中一个接口的入参依赖另一个接口的结果，本想让前端给我传过来，但是这样的话又有一个接口会分离出去，影响首页加载速度，于是决定还是我后端想办法处理；如此一来，就需要解决查询时的前后依赖关系，一开始想了几种方案：

使用CountDownLatch进行等待
拿到前一个依赖任务的Future，调用get方法等待
有依赖的任务串行执行

前两种方案大同小异，其实都是利用线程间的等待机制，待前置查询执行完成后，唤醒当前线程继续执行查询；第三种方案则简单粗暴，由于时间较紧，经过短暂思考后权利利弊还是决定用第三种方案，毕竟我现在只有两个子查询是有依赖的，这样做也是效率最高的办法；于是在生成查询的supplier中做了下手脚，将有依赖关系的Function通过andThen方法聚合为一个Function，便完成了需求，这样子查询顺序变成了（假设B依赖A）：

至此，在需求上来说已经实现完成了，但是从技术角度来说不能算完整：假如新来了一个接口G，G也依赖于A，那么按照这种方案，ABG将会串行执行，然而BG之间没有依赖关系，它俩理应并行；又或者B依赖于A和C，则ABC会串行执行，然而AC应该并行才合理；再极端的例子，B依赖了ACEF，则整个聚合查询都退化为了单线程，显然更不合理。

所以，虽然现阶段这种方案没什么问题，但对于复杂的依赖关系处理是远不够完善的，因此我又思考了有没有更好的方式去处理，并且能够有一定的通用性。

重构的并行任务处理方案

在此类需求中，其实比较重要的一点就是如何处理这些任务的依赖关系，而且尽可能的让他们并行处理，由此引入了一些图的应用。在我的方案中，有以下几个概念：

ExecutionContext：执行的上下文环境，所有子任务共用，用于处理数据交换和输入输出的问题，其实本质上是一个线程安全的Map。
ExecutionNode：任务节点，即一个子任务的抽象，内部包含：名称、依赖节点名称、可执行的Function
ExecutionGraph：整个任务流的执行图，包含所有的执行节点以及排序后的节点，任务流启动时将按照此类的描述执行。
ExecutionFlow：任务流的抽象，本质上也是一个Function，调用apply方法开始执行。
FlowConfiguration：任务执行的一些配置，包含前置Action、后置Action、线程执行器、超时时间等。

图的生成

ExecutionGraph是将任务的执行过程抽象成了一个有向无环图，在使用时，传入每个节点的定义及提前定义好的执行过程(Function)即可生成一个图。

其中节点的定义规则为：{节点名称}[{依赖节点A},{依赖节点B},...]

若无依赖，则没有方括号中的内容。具体在构建执行图时，则相当于是对这个有向图做了一个简单的排序（拓扑排序）：

首先将所有节点加入到remainMap，preMap为空
找到没有依赖的节点，作为第一层，并将这些节点从remainMap移除，加入到preMap中
遍历remainMap，找到所有依赖节点均已经在preMap中出现的节点，加入到preMap，并在remainMap移除，这些元素作为下一层
循环第3步，直至remainMap为空

需要注意的是，第3步中，如果remainMap在筛选前后大小不变，则认为是出现了循环依赖，或者依赖了一个不存在的节点，即有向图中出现了环，应抛出异常，否则将陷入死循环。核心代码如下：

 /** * 构建执行图，实质上是对节点做排序 * * @param preNodes 已处理的节点 * @param remainNode 剩余节点 * @param levels 有序节点（结果集） */ private static void buildGraph(Map<String, ExecutionNode> preNodes, Map<String, ExecutionNode> remainNode, List<List<ExecutionNode>> levels) { // 若剩余节点为空，结束 if (remainNode.isEmpty()) { return; } List<String> currentLevel; // 记录剩余节点个数 int remainSize = remainNode.size(); if (preNodes.isEmpty()) { // 若preNodes为空，则为第一层，找出无依赖的节点，即入度为0的节点 currentLevel = remainNode.values() .stream() .filter(e -> e.getDepend().length == 0) .map(ExecutionNode::getName) .collect(Collectors.toList()); // 若没有无依赖的节点，抛出异常 if (currentLevel.isEmpty()) { throw new IllegalStateException("could not found init node"); } } else { // 继续寻找前置节点已处理的节点，相当于继续寻找删除前面的节点依赖后，入度为0的节点 currentLevel = remainNode.entrySet() .stream() .filter(e -> Stream.of(e.getValue().getDepend()).allMatch(preNodes::containsKey)) .map(Map.Entry::getKey) .collect(Collectors.toList()); } List<ExecutionNode> nodes = new ArrayList<>(currentLevel.size()); // currentLevel中为当前层级的节点，将其加入preNodes中记录起来，并从剩余节点中移除 for (String name : currentLevel) { nodes.add(remainNode.get(name)); preNodes.put(name, remainNode.remove(name)); } // 若为剩余节点数量不变，说明图中有环形结构，或依赖了未定义的节点 if (remainNode.size() == remainSize) { throw new IllegalStateException("there may be circular dependencies or dependency does not exist in your graph, exception nodes is: " + remainNode.values()); } // 放入结果集 levels.add(nodes); // 递归至下一层 buildGraph(preNodes, remainNode, levels); }

示例简单介绍个例子，对于如下节点定义：

A B C D E[A,B] F[A,E] G[C] H[B,C] I[G,H] J[I] K[D,E] L[E,H]

则对应的图为：

这样，所有的执行任务被分为了四层，每一层之间的任务是互不依赖的，同时下一层的执行依赖上一层。

任务节点执行方案

按照这种图的排序算法分割之后，剩下的就比较简单了，我能想到的有两种方式：

每一层并行执行，全部完成之后执行下一层
利用线程间等待机制，等待前置任务执行完成通知当前线程

第一种我觉得不算很好，例如上图，考虑如下情况：L的执行时间较长，大于I和J的执行时间总和，即 t(L) > t(I) + t(J)，则J完全没必要等待第三层全部执行完成，当I结束之后J就可以执行，若t(J) > t(L)时，更是浪费了较多时间。所以我采用了CompleteFuture的get方法阻塞有依赖的线程，只要前置任务执行完毕，当前任务就可以开始，而不必等待上一层所有的任务执行完毕。

方案如下：

遍历每一层节点

a. 对于当前层级的每一个节点，生成CompleteFuture，在构造Supplier时，先从futureMap获取所有的依赖节点的Future，循环调用其get方法，阻塞当前线程（若已经执行完成则直接返回结果，此处也可以使用allOf.get()方法代替）

b. 将构造完成的CompleteFuture加入futureMap中
将futureMap所有的CompleteFuture取出来，使用allOf(...).get()方法等待全部执行完成。

理论上每一个节点只要前置执行完成，它就会开始，但是实际上，由于线程池调度等原因，可能还会在阻塞队列中等待一段时间，但是站在系统资源分配的角度，可以接受其带来的影响。实现的核心代码：（其实这种传入参数并生成新函数的方式还有一个高大上的名字：函数柯里化？=_= ）

 /** * 生成一个执行任务，返回的是一个新函数 * * @param executionGraph 执行任务图 * @param flowConfiguration 执行参数 * @return */ static ExecutionFlow buildFlow(@NonNull ExecutionGraph executionGraph, @NonNull FlowConfiguration flowConfiguration) { SpelExpressionParser expressionParser = new SpelExpressionParser(); return ctx -> { long startTime = System.currentTimeMillis(); List<List<ExecutionNode>> sortedNodes = executionGraph.getSortedNodes(); Map<String, CompletableFuture<ExecutionContext>> futureMap = new HashMap<>(16); // 按图序构建CompleteFuture for (List<ExecutionNode> nodes : sortedNodes) { // 同一层级之间的节点没有依赖，且依赖的CompletableFuture一定已经构建完成 for (ExecutionNode node : nodes) { String currNodeName = node.getName(); CompletableFuture<ExecutionContext> future = CompletableFuture.supplyAsync(() -> { try { long t1 = System.currentTimeMillis(); // 前置操作 ofNullable(flowConfiguration.getPreAction()).ifPresent(c -> c.accept(ctx)); String[] depend = node.getDepend(); // 遍历依赖节点，确保依赖节点已经执行完成 for (String s : depend) { try { // 阻塞等待 futureMap.get(s).get(); logger.info("current node [{}] waiting node [{}] done ({}ms)", currNodeName, s, System.currentTimeMillis() - t1); } catch (InterruptedException | ExecutionException e) { throw new IllegalStateException("current node [" + currNodeName + "] waiting node [" + s + "] failed", e); } } long t2 = System.currentTimeMillis(); logger.info("current node [{}] start", currNodeName); // 前置表达式参数检查 //noinspection ConstantConditions if (Objects.nonNull(node.getBeforeCheckExpression()) && !node.getBeforeCheckExpression().getValue(ctx.getCtx(), boolean.class)) { throw new IllegalStateException(String.format("before check failed by expression [%s]" , node.getBeforeCheckExpression().getExpressionString())); } // 执行节点 ExecutionContext ret = node.getExecution().apply(ctx); // 用于标识节点是否正确执行 ctx.setVar("$" + currNodeName + "_success", true); // 后置表达式参数检查 //noinspection ConstantConditions if (Objects.nonNull(node.getAfterCheckExpression()) && !node.getAfterCheckExpression().getValue(ctx.getCtx(), boolean.class)) { throw new IllegalStateException(String.format("after check failed by expression [%s]" , node.getAfterCheckExpression().getExpressionString())); } long tl = System.currentTimeMillis(); logger.info("current node [{}] execute done (total: {}ms, execution: {}ms)", currNodeName, tl - t1, tl - t2); return ret; } catch (Throwable e) { logger.warn("current node [{}] execute failed: {}, skip exception={}", currNodeName, e.getMessage(), node.isSkipOnFail(), e); if (node.isSkipOnFail()) { return ctx; } throw e; } finally { // 后置操作 ofNullable(flowConfiguration.getFinalAction()).ifPresent(c -> c.accept(ctx)); } }, flowConfiguration.getExecutor() == null ? ForkJoinPool.commonPool() : flowConfiguration.getExecutor()); // 将构建好的CompletableFuture放入Map futureMap.put(node.getName(), future); } } try { // 组合所有的Future，等待所有节点行完毕 CompletableFuture<Void> completableFuture = CompletableFuture.allOf(futureMap.values().toArray(new CompletableFuture[0])); logger.info("waiting flow execution down, ttl={}", flowConfiguration.getTimeout()); if (flowConfiguration.getTimeout() > 0) { completableFuture.get(flowConfiguration.getTimeout(), TimeUnit.SECONDS); } else { completableFuture.get(); } logger.info("execution flow success ({}ms)", System.currentTimeMillis() - startTime); return ctx; } catch (InterruptedException | ExecutionException | TimeoutException e) { logger.warn("execution flow failed: {} ({}ms)", e.getMessage(), System.currentTimeMillis() - startTime); throw new IllegalStateException(e); } }; }

扩展功能

虽然已经完成了我们需要的功能，但是还不够完善，例如在实际执行每一个节点时，有的节点可以容忍失败，有的不可以，或者想在节点执行前后能检查一些参数是否满足要求，等等，所以我在这里又对节点定义的表达式做了一些增强，以此来支持一些额外的操作，同时也能在后续进行扩展：

在原有的基础上做一下改造：

{节点名称}[{依赖节点A},{依赖节点B}, ...]({扩展操作A名称}({表达式}), {扩展操作A名称}({表达式}), ...)

是否忽略失败

用于在当前节点执行发生异常的时候，是否将异常抛出，还是忽略这个异常

扩展名称：skipOnFail

示例：E[A,B](skipOnFail(true))

...... } catch (Throwable e) { log.warn("current node [{}] execute failed: {}, skip exception={}", currNodeName, e.getMessage(), node.isSkipOnFail(), e); if (node.isSkipOnFail()) { return ctx; } throw e; } finally { ......

参数检查

在针对有依赖的场景下，当前节点如何知道所依赖的节点是否执行正常呢，若有异常，我们可以通过异常得知执行失败，但是有时候不会有异常抛出（例如调用的子方法有拦截器的时候，有可能在拦截器中会处理掉异常信息），需要检查数据才能得知是否正确执行，如果把这个检查的操作由节点内部定义的函数来做，似乎是比较繁琐的，而且这一部分我觉得由框架来统一处理掉是比较合适的，我想到的方案是用表达式的形式来检查参数：给每个节点增加一个前置和后置的表达式，表达式的返回值是boolean类型，若为false则抛出异常；

表达式结构：{节点名称}[{依赖节点A},{依赖节点B}, ...](beforeCheck({布尔表达式}), afterCheck({布尔表达式}))

示例：J[I](beforeCheck([c] == 0 && [k] == 9 && [c] != null), afterCheck([a] == 1))

表达式遵循SpringEL语法，解析时的上下文为ExecutionContext中的变量

执行样例

执行样例代码：

public static void main(String[] args) { Random random = new Random(); // 简单映射一个函数集合，随机暂停一段时间，模拟函数执行，部分函数里面做了一些简单的值的设定，模拟结果输出 Map<String, Function<ExecutionContext, ExecutionContext>> functionMap = Arrays.stream("A,B,C,D,E,F,G,H,I,J,K,L".split(",")).collect(Collectors.toMap( e -> e, e -> ctx -> { try { int i = 500 + 500 * random.nextInt(5); Thread.sleep(i); ctx.setVar(e, i); switch (e) { case "A": ctx.setVar("a", 1); break; case "B": ctx.setVar("a", 0); break; case "C": ctx.setVar("c", 3); break; case "G": ctx.setVar("g", 9); ctx.setVar("a", 1); break; } } catch (InterruptedException ignored) { } return ctx; } )); // 构造一个执行图的定义 ExecutionGraph graph = ExecutionGraph.createGraph(new ArrayList<>() {{ add("A"); add("B[A](beforeCheck([a] == 1))"); add("C"); add("D"); add("E[A,B](beforeCheck([a]==2); skipOnFail(true))"); // 该节点会发生异常，但是配置了跳过 add("F[A,E]"); add("G[C]"); add("H[B,C]"); add("I[G,H]"); add("J[I](beforeCheck( T(java.util.Arrays).asList(1,2,3).contains([c]) and [g] == [c]*[c] and [c] != null); afterCheck([c] >= 1))"); add("K[D,E]"); add("L[E,H]"); }}, functionMap); // CompletableFuture默认使用的ForkJoinPool.commonPool，可以试下使用cachedThreadPool会有什么不一样？ // ExecutorService executorService = Executors.newCachedThreadPool(); ExecutionFlow executionFlow = ExecutionFlow.buildFlow(graph); executionFlow.apply(new ExecutionContext()); }

执行日志：

/Library/Java/JavaVirtualMachines/jdk-11.0.8.jdk/Contents/Home/bin/java -Dvisualvm.id=76728482855487 -javaagent:/Users/windlively/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/213.6461.79/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=55008:/Users/windlively/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/213.6461.79/IntelliJ IDEA.app/Contents/bin -Dfile.encoding=UTF-8 -classpath /Users/windlively/IdeaProjects/leisure/frame/target/classes:/Users/windlively/.m2/repository/org/projectlombok/lombok/1.18.10/lombok-1.18.10.jar:/Users/windlively/.m2/repository/org/springframework/spring-expression/5.2.2.RELEASE/spring-expression-5.2.2.RELEASE.jar:/Users/windlively/.m2/repository/org/springframework/spring-core/5.2.2.RELEASE/spring-core-5.2.2.RELEASE.jar:/Users/windlively/.m2/repository/org/springframework/spring-jcl/5.2.2.RELEASE/spring-jcl-5.2.2.RELEASE.jar:/Users/windlively/.m2/repository/org/springframework/boot/spring-boot-starter-logging/2.2.2.RELEASE/spring-boot-starter-logging-2.2.2.RELEASE.jar:/Users/windlively/.m2/repository/ch/qos/logback/logback-classic/1.2.3/logback-classic-1.2.3.jar:/Users/windlively/.m2/repository/ch/qos/logback/logback-core/1.2.3/logback-core-1.2.3.jar:/Users/windlively/.m2/repository/org/slf4j/slf4j-api/1.7.29/slf4j-api-1.7.29.jar:/Users/windlively/.m2/repository/org/apache/logging/log4j/log4j-to-slf4j/2.12.1/log4j-to-slf4j-2.12.1.jar:/Users/windlively/.m2/repository/org/apache/logging/log4j/log4j-api/2.12.1/log4j-api-2.12.1.jar:/Users/windlively/.m2/repository/org/slf4j/jul-to-slf4j/1.7.29/jul-to-slf4j-1.7.29.jar:/Users/windlively/.m2/repository/org/apache/commons/commons-lang3/3.9/commons-lang3-3.9.jar ink.windlively.frame.parallelprocessor.Main 12:38:04.122 [main] INFO ink.windlively.frame.parallelprocessor.ExecutionGraph - start create graph, node definition: [A, B[A<-](beforeCheck([a] == 1)), C, D, E[A<-,B<-](beforeCheck([a]==2),skipOnFail(true)), F[A<-,E<-], G[C<-], H[B<-,C<-], I[G<-,H<-], J[I<-](beforeCheck(T(java.util.Arrays).asList(1,2,3).contains([c]) and [g] == [c]*[c] and [c] != null),afterCheck([c] >= 1)), K[D<-,E<-], L[E<-,H<-]] 12:38:04.165 [main] INFO ink.windlively.frame.parallelprocessor.ExecutionGraph - build execution graph success: A C D B[A<-](beforeCheck([a] == 1)) G[C<-] E[A<-,B<-](beforeCheck([a]==2),skipOnFail(true)) H[B<-,C<-] F[A<-,E<-] I[G<-,H<-] K[D<-,E<-] L[E<-,H<-] J[I<-](beforeCheck(T(java.util.Arrays).asList(1,2,3).contains([c]) and [g] == [c]*[c] and [c] != null),afterCheck([c] >= 1)) 12:38:04.183 [main] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - waiting flow execution down, ttl=0 12:38:04.183 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [C] start 12:38:04.184 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [A] start 12:38:04.183 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [D] start 12:38:05.189 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [D] execute done (total: 1006ms, execution: 1006ms) 12:38:05.687 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [A] execute done (total: 1504ms, execution: 1503ms) 12:38:05.689 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [F] waiting node [A] done (498ms) 12:38:05.689 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [B] waiting node [A] done (1506ms) 12:38:05.689 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [E] waiting node [A] done (1505ms) 12:38:05.690 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [B] start 12:38:06.688 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [C] execute done (total: 2505ms, execution: 2505ms) 12:38:06.689 [ForkJoinPool.commonPool-worker-11] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [G] waiting node [C] done (2505ms) 12:38:06.689 [ForkJoinPool.commonPool-worker-11] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [G] start 12:38:06.689 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [K] waiting node [D] done (0ms) 12:38:07.247 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [B] execute done (total: 3064ms, execution: 1557ms) 12:38:07.248 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [E] waiting node [B] done (3064ms) 12:38:07.248 [ForkJoinPool.commonPool-worker-13] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [H] waiting node [B] done (3064ms) 12:38:07.248 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [E] start 12:38:07.248 [ForkJoinPool.commonPool-worker-13] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [H] waiting node [C] done (3064ms) 12:38:07.248 [ForkJoinPool.commonPool-worker-13] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [H] start 12:38:07.261 [ForkJoinPool.commonPool-worker-15] WARN ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [E] execute failed: before check failed by expression [[a]==2], skip exception=true java.lang.IllegalStateException: before check failed by expression [[a]==2] at ink.windlively.frame.parallelprocessor.ExecutionFlow.lambda$buildFlow$2(ExecutionFlow.java:81) at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700) at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1692) at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290) at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020) at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656) at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594) at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183) 12:38:07.262 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [F] waiting node [E] done (2071ms) 12:38:07.262 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [F] start 12:38:07.262 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [K] waiting node [E] done (573ms) 12:38:07.262 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [L] waiting node [E] done (14ms) 12:38:07.262 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [K] start 12:38:07.767 [ForkJoinPool.commonPool-worker-9] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [F] execute done (total: 2576ms, execution: 505ms) 12:38:08.194 [ForkJoinPool.commonPool-worker-11] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [G] execute done (total: 4010ms, execution: 1505ms) 12:38:08.195 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [I] waiting node [G] done (2506ms) 12:38:09.266 [ForkJoinPool.commonPool-worker-5] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [K] execute done (total: 2577ms, execution: 2004ms) 12:38:09.749 [ForkJoinPool.commonPool-worker-13] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [H] execute done (total: 5565ms, execution: 2501ms) 12:38:09.750 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [L] waiting node [H] done (2502ms) 12:38:09.750 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [I] waiting node [H] done (4062ms) 12:38:09.751 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [L] start 12:38:09.751 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [I] start 12:38:10.753 [ForkJoinPool.commonPool-worker-7] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [L] execute done (total: 3505ms, execution: 1002ms) 12:38:11.752 [ForkJoinPool.commonPool-worker-3] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [I] execute done (total: 6064ms, execution: 2001ms) 12:38:11.753 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [J] waiting node [I] done (4491ms) 12:38:11.753 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [J] start 12:38:12.301 [ForkJoinPool.commonPool-worker-15] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - current node [J] execute done (total: 5039ms, execution: 548ms) 12:38:12.301 [main] INFO ink.windlively.frame.parallelprocessor.ExecutionFlow - execution flow success (8130ms) 进程已结束,退出代码0

后记

在引入SpringEL表达式之后，其扩展性已经非常强了，Spring表达式支持Java的许多语法、Spring Bean的调用、方法调用等，例如可以仅用配置表达式的方式，在表达式中通过Bean调用的方式定义任务执行过程，并自动设置出入参，配置出一个任务流程，而无需再在代码中预先定义好functionMap。

该小框架已经上线稳定运行一段时间了，支撑着我们业务APP首页数据大量的查询请求，将原有几秒钟的接口总耗时将至平均300ms左右，但仍有许多可以改进的地方，也欢迎提出意见，共同探讨。