Вложение CompletionStages в Java для запуска внутренних блоков перед внешними блоками

Question

Я написал метод, подобный следующему:

 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, пожалуйста, посоветуйте мне, как убедиться, что внутренний блок завершается довнешний блок?

Answer 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); })