数据结构与算法的实际应用（有向图）——有依赖的任务并行处理框架
背景
一开始接到一个技改需求，需要将我们系统的一些首页查询接口进行优化，合并为一个聚合接口提供给前端；最初与前端同学进行交流，认为这些首页查询接口没有依赖关系，因此对于我来说解决方案比较简单，直接将需要查询的接口进行并行处理即可：
其中每一个查询抽象成了一个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左右，但仍有许多可以改进的地方，也欢迎提出意见，共同探讨。
