在 Java 中嵌套 CompletionStages 以使内部块在外部块之前运行答案

【问题标题】：Nesting CompletionStages in Java to make inner blocks run before outer blocks在 Java 中嵌套 CompletionStages 以使内部块在外部块之前运行
【发布时间】：2018-10-19 06:40:04
【问题描述】：

我写了一个方法如下：

 public static CompletionStage<Tuple2<ObjectNode, String>> calculateTemplateTreeAndKeys(
  String content,
  RequestContext context,
  MetricsClient metricsClient,
  JdbcSession jdbcSession) {

AtomicReference<ObjectNode> templateTreeHolder = new AtomicReference<>();
templateTreeHolder.set(Json.rootNode());

return getTemplateIds(context, metricsClient, jdbcSession, content)
    .thenCompose(
        templateIds -> {
          templateIds.map(
              id ->
                  // do something and return CompletionStage<String>
                      .thenAccept(
                          tree -> {
                            templateTreeHolder.set(
                                (ObjectNode)
                                    templateTreeHolder.get().set(id, Json.readTree(tree)));

                            System.out.println(
                                "From inner function: " + templateTreeHolder.get());
                          }));
          return CompletableFuture.completedFuture(NotUsed.getInstance());
        })
    .thenApply(
        notUsed -> {
          String includedTemplateIdsStr =
              getKeysFromTemplateTree(templateTreeHolder.get()).toJavaList().toString();

          System.out.println("From outer function: " + templateTreeHolder.get());

          return Tuple.of(templateTreeHolder.get(), includedTemplateIdsStr);
        });

我希望内部块在之前处理和更新 templateTreeHolder .thenApply 被调用，因此 templateTreeHolder 将保存正确的数据以返回。但是，.thenApply 块在内部 .thenAccept 块之前处理。

从控制台输出序列：

From outer function: {}
From inner function: {"f9406341-c62a-411a-9389-00a62bd63629":{}}

我不确定我在链接 CompletionStages 时做错了什么，请告诉我如何确保内部块在外部块之前完成？

【问题讨论】：

首先，我认为你的例子太复杂了。其次，你没有使用任何async 方法，为什么你甚至需要像这样进行链接？
@Jai 该项目基于事件溯源/CQRS，我们需要确保每个事件在下一个之前处理，因此有意避免任何异步方法。我对为什么 .thenApply 在 .thenAccept 之前运行以及如何避免这种情况感到困惑。
如果他们需要顺序运行，那么只需删除内部的CompletableFuture。那不是更容易吗？我想不出任何理由 CompletableFuture 会在非异步上下文中做任何有用的事情。
@Jai 感谢您的建议，但目前不可能，代码库与基于 Lagom、akka 持久性、Vavr 库的 CompletionStages 捆绑在一起。无论如何，如果你能就这个确切的问题提出一些建议，那将会很有帮助。
你得到你想要的。当CompletableFuture.completedFuture(NotUsed.getInstance()) 完成时，评估最后一个函数。我们无法知道传递给 templateIds.map(…) 的函数何时被评估，因为它与您的其余代码（尤其是完成阶段链）完全无关。

标签： java multithreading asynchronous java-8 java.util.concurrent

【解决方案1】：

您传递给thenCompose 的函数正在返回一个已经完成的未来，即return CompletableFuture.completedFuture(NotUsed.getInstance());，它允许相关阶段立即进行。这似乎与传递给templateIds.map(…) 的函数的评估相冲突，这显然是异步发生的。

通常，您应该避免将完成阶段和对副作用的依赖关系混合在一起，尤其是当它们的异步评估未建模为先决条件完成阶段时。

但如果您别无选择，您可以解决此问题：

return getTemplateIds(context, metricsClient, jdbcSession, content)
    .thenCompose(
        templateIds -> {
          // create an initially uncompleted stage
          CompletableFuture<Object> subStage = new CompletableFuture<>(); 
          templateIds.map(
              id ->
                  // do something and return CompletionStage<String>
                      .thenAccept(
                          tree -> {
                            templateTreeHolder.set(
                                (ObjectNode)
                                    templateTreeHolder.get().set(id, Json.readTree(tree)));

                            System.out.println(
                                "From inner function: " + templateTreeHolder.get());
                            // complete when all work has been done
                            subStage.complete(null);
                          }));
          // use this stage for dependent actions
          return subStage;
        })
    .thenApply(
        notUsed -> {
          String includedTemplateIdsStr =
              getKeysFromTemplateTree(templateTreeHolder.get()).toJavaList().toString();

          System.out.println("From outer function: " + templateTreeHolder.get());

          return Tuple.of(templateTreeHolder.get(), includedTemplateIdsStr);
        });

在上面的代码中，如果您的操作在完成尝试之前因异常而失败，则未来将永远不会完成。一般模式是这样的：

CompletableFuture<Type> stage = new CompletableFuture<>();
…
try {
    code that will eventually call complete on stage
}
catch(Throwable t) {
    stage.completeExceptionally(t);
}

但是，当然，当应该完成阶段的代码也承担异步处理时，它会变得有点复杂，所以你必须保护试图提交实际完成代码的代码，以及实际完成代码.

因此，更详细的内部代码版本如下所示：

CompletableFuture<Object> subStage = new CompletableFuture<>();
try {
    templateIds.map(
        id ->
            // do something and return CompletionStage<String>
            .thenAccept(
                tree -> {
                  templateTreeHolder.set(
                      (ObjectNode)
                          templateTreeHolder.get().set(id, Json.readTree(tree)));

                  System.out.println(
                      "From inner function: " + templateTreeHolder.get());
                })
            .whenComplete((v,t) -> {
                // complete when all work has been done
                if(t != null) subStage.completeExceptionally(t);
                else subStage.complete(v);
            }));
} catch(Throwable t) {
    subStage.completeExceptionally(t);
}
// use this stage for dependent actions
return subStage;

（也许“做某事并返回CompletionStage”也必须用try { … } catch(Throwable t) { subStage.completeExceptionally(t); }保护）

【讨论】：