将 CompletableFutures 组合成树状结构

Question

我目前正在试验 Java 8 的 Future API 和 Core CompleteableFuture。我知道它是一个 monad，所以它应该有一个 bind-Operator。

现在的想法是我有一个抽象操作 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()))。

同样适用于两个CompletableFutures 和一个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();