java - 将 CompletableFutures 组合成树状结构

Question

我目前正在试验 Java 8 的 Future API 和 Core CompleteableFuture。我知道它是一个 monad，所以它应该有一个绑定(bind)操作符。

现在的想法是我有一棵抽象操作 k 的树，例如像这样:

O --- > O3 ---> O4 ---> o5
  O ---> o2 --/

我想将 O3 和 O2“合并”到一个新的操作中 O4 = (O3 o O2) (t) = O2(O3(t) )。我的想法是我是例如在 O2 中，想说:将当前节点与 O3 合并并返回一个由串联操作组成的新节点。不幸的是，我已经尝试了一整晚了，但我无法弄清楚。此外，出于未知原因，使用诸如 O1.mergeWith(O2).mergeWith(O3) 之类的运算符会在单次调用中触发 apply 方法两次。

目标是创建一个由其他函数组成的新函数，这样我就可以尽可能推迟计算。

public abstract class Operation<T,R> {
  // The value a
  private final CompletableFuture<R> future;

// Executes the operation on t and returns something (maybe different) of type R
  protected abstract R apply(T t);

// Gets the value of the future of this operation
  public R get() throws Exception {
    return future.get();
  }

  protected Operation(Supplier<T> s) {
    future =  CompletableFuture.supplyAsync(s).thenApplyAsync(transform());
  }

  protected Operation(CompletableFuture<R> f) {
    future = f;
  }


  CompletableFuture<R> createTargetFuture(R _value) {
    return CompletableFuture.supplyAsync(() -> _value);
  }

  <S> Operation<T,S> mergeWith(Operation<R,S> _other) {
    CompletableFuture<S> _result = future.thenComposeAsync(
      // Method is synchronous, but is run async :) We need the createTargetFuture,         otherwise we cannot turn the constant t into a producer.
      (t) -> _other.createTargetFuture(_other.apply(t)));
      //transform()).thenApplyAsync(_other.transform());
    return new Operation<T, S>(_result) {
      @Override
      protected S apply(T t) {
        // What to put here
        return null;
      }
    };
  }

  protected Function<T,R> transform(){
    return this::apply;
  }
}

Answer 1

目前还不清楚您想要实现什么目标。您的整个Operation类看起来您正在向 CompletableFuture 添加功能它已经自己提供了。

如果您有 Supplier<T> s想要推迟其执行，你可以这样做 CompletableFuture.supplyAsync(s)正如你已经知道的。

现在，如果您想申请Function<T,R> f到 CompletableFuture<T> a 的延迟结果只需使用 CompletableFuture.supplyAsync(()->f.apply(a.join())) .

对于两个 CompletableFuture 也同样有效和a BiFunction :CompletableFuture.supplyAsync(()->f.apply(a.join(),b.join()))

如果您觉得需要将操作包装在某种支持代码中，您可以创建如下所示的实用程序方法:

public static <T> CompletableFuture<T> create(Supplier<T> s) {
    return CompletableFuture.supplyAsync(s);
}
public static <T,R> CompletableFuture<R> create(
  Function<T,R> f, CompletableFuture<T> a) {
    return CompletableFuture.supplyAsync(()->f.apply(a.join()));
}
public static <T,U,R> CompletableFuture<R> create(
  BiFunction<T,U,R> f, CompletableFuture<T> a, CompletableFuture<U> b) {
    return CompletableFuture.supplyAsync(()->f.apply(a.join(),b.join()));
}

然后您就可以愉快地执行异步操作，例如:

int i=create((a,b)->a+b, create(()->42),create(()->100)).join();

但是，当然，同样的事情也可以在没有任何支持方法的情况下工作:

int i=CompletableFuture.supplyAsync(() ->
    CompletableFuture.supplyAsync(()->42).join()
  + CompletableFuture.supplyAsync(()->100).join() ).join();

或

int j=CompletableFuture.supplyAsync(()->42).thenApplyAsync(
    a-> a + CompletableFuture.supplyAsync(()->100).join() ).join();