将 Java Future 转换为 CompletableFuture

Question

Java 8 引入了CompletableFuture，这是一个可组合的 Future 新实现（包括一堆 thenXxx 方法）。我想专门使用它，但我想使用的许多库只返回不可组合的 Future 实例。

有没有办法将返回的 Future 实例包装在 CompleteableFuture 中，以便我编写它？

Answer 1

如果您要使用的库除了 Future 样式之外还提供回调样式方法，您可以为其提供一个处理程序，该处理程序完成 CompletableFuture 而无需任何额外的线程阻塞。像这样：

    AsynchronousFileChannel open = AsynchronousFileChannel.open(Paths.get("/some/file"));
    // ... 
    CompletableFuture<ByteBuffer> completableFuture = new CompletableFuture<ByteBuffer>();
    open.read(buffer, position, null, new CompletionHandler<Integer, Void>() {
        @Override
        public void completed(Integer result, Void attachment) {
            completableFuture.complete(buffer);
        }

        @Override
        public void failed(Throwable exc, Void attachment) {
            completableFuture.completeExceptionally(exc);
        }
    });
    completableFuture.thenApply(...)

如果没有回调，我认为解决此问题的唯一其他方法是使用轮询循环，将所有 Future.isDone() 检查放在单个线程上，然后在 Future 可获取时调用完成。

Answer 2

有一种方法，但你不会喜欢它。以下方法将Future<T> 转换为CompletableFuture<T>：

public static <T> CompletableFuture<T> makeCompletableFuture(Future<T> future) {
  if (future.isDone())
    return transformDoneFuture(future);
  return CompletableFuture.supplyAsync(() -> {
    try {
      if (!future.isDone())
        awaitFutureIsDoneInForkJoinPool(future);
      return future.get();
    } catch (ExecutionException e) {
      throw new RuntimeException(e);
    } catch (InterruptedException e) {
      // Normally, this should never happen inside ForkJoinPool
      Thread.currentThread().interrupt();
      // Add the following statement if the future doesn't have side effects
      // future.cancel(true);
      throw new RuntimeException(e);
    }
  });
}

private static <T> CompletableFuture<T> transformDoneFuture(Future<T> future) {
  CompletableFuture<T> cf = new CompletableFuture<>();
  T result;
  try {
    result = future.get();
  } catch (Throwable ex) {
    cf.completeExceptionally(ex);
    return cf;
  }
  cf.complete(result);
  return cf;
}

private static void awaitFutureIsDoneInForkJoinPool(Future<?> future)
    throws InterruptedException {
  ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    @Override public boolean block() throws InterruptedException {
      try {
        future.get();
      } catch (ExecutionException e) {
        throw new RuntimeException(e);
      }
      return true;
    }
    @Override public boolean isReleasable() {
      return future.isDone();
    }
  });
}

显然，这种方法的问题在于，对于每个 Future，一个线程将被阻塞以等待 Future 的结果——与期货。在某些情况下，可能会做得更好。但是，一般来说，不主动等待Future的结果是没有解决办法的。

Answer 3

如果您的Future 是调用ExecutorService 方法的结果（例如submit()），最简单的方法是改用CompletableFuture.runAsync(Runnable, Executor) 方法。

来自

Runnbale myTask = ... ;
Future<?> future = myExecutor.submit(myTask);

到

Runnbale myTask = ... ;
CompletableFuture<?> future = CompletableFuture.runAsync(myTask, myExecutor);

然后“本地”创建 CompletableFuture。

编辑：@SamMefford 追求 cmets 由 @MartinAndersson 纠正，如果你想传递 Callable，你需要调用 supplyAsync()，将 Callable<T> 转换为 Supplier<T>，例如与：

CompletableFuture.supplyAsync(() -> {
    try { return myCallable.call(); }
    catch (Exception ex) { throw new CompletionException(ex); } // Or return default value
}, myExecutor);

因为T Callable.call() throws Exception; 会引发异常而T Supplier.get(); 不会引发异常，所以您必须捕获异常以便原型兼容。

异常处理说明

get() 方法没有指定throws，这意味着它不应该抛出 checked 异常。但是，可以使用 unchecked 异常。 CompletableFuture 中的代码显示 CompletionException 已使用且未选中（即是 RuntimeException），因此捕获/抛出将任何异常包装到 CompletionException 中。

另外，正如@WeGa 所指出的，您可以使用handle() 方法来处理结果可能引发的异常：

CompletableFuture<T> future = CompletableFuture.supplyAsync(...);
future.handle((ex,res) -> {
        if (ex == null) {
            // An exception occurred ...
        } else {
            // No exception was thrown, 'res' is valid and can be handled here
        }
    });

Answer 4

我发布了一个小的futurity 项目，试图在答案中做得比the straightforward way 更好。

主要思想是使用唯一的一个线程（当然不仅仅是一个自旋循环）来检查内部的所有 Futures 状态，这有助于避免为每个 Future -> CompletableFuture 转换阻塞池中的线程。

使用示例：

Future oldFuture = ...;
CompletableFuture profit = Futurity.shift(oldFuture);

Answer 5

建议：

http://www.thedevpiece.com/converting-old-java-future-to-completablefuture/

但是，基本上：

public class CompletablePromiseContext {
    private static final ScheduledExecutorService SERVICE = Executors.newSingleThreadScheduledExecutor();

    public static void schedule(Runnable r) {
        SERVICE.schedule(r, 1, TimeUnit.MILLISECONDS);
    }
}

还有，CompletablePromise：

public class CompletablePromise<V> extends CompletableFuture<V> {
    private Future<V> future;

    public CompletablePromise(Future<V> future) {
        this.future = future;
        CompletablePromiseContext.schedule(this::tryToComplete);
    }

    private void tryToComplete() {
        if (future.isDone()) {
            try {
                complete(future.get());
            } catch (InterruptedException e) {
                completeExceptionally(e);
            } catch (ExecutionException e) {
                completeExceptionally(e.getCause());
            }
            return;
        }

        if (future.isCancelled()) {
            cancel(true);
            return;
        }

        CompletablePromiseContext.schedule(this::tryToComplete);
    }
}

例子：

public class Main {
    public static void main(String[] args) {
        final ExecutorService service = Executors.newSingleThreadExecutor();
        final Future<String> stringFuture = service.submit(() -> "success");
        final CompletableFuture<String> completableFuture = new CompletablePromise<>(stringFuture);

        completableFuture.whenComplete((result, failure) -> {
            System.out.println(result);
        });
    }
}

Answer 6

让我建议另一个（希望是更好的）选项： https://github.com/vsilaev/java-async-await/tree/master/com.farata.lang.async.examples/src/main/java/com/farata/concurrent

简单来说，思路如下：

1. 引入CompletableTask<V>接口——联合 CompletionStage<V> + RunnableFuture<V>
2. 将ExecutorService 变形为从submit(...) 方法返回CompletableTask（而不是Future<V>）
3. 完成，我们拥有可运行和可组合的 Future。

实现使用替代的 CompletionStage 实现（注意，CompletionStage 而不是 CompletableFuture）：

用法：

J8ExecutorService exec = J8Executors.newCachedThreadPool();
CompletionStage<String> = exec
   .submit( someCallableA )
   .thenCombineAsync( exec.submit(someCallableB), (a, b) -> a + " " + b)
   .thenCombine( exec.submit(someCallableC), (ab, b) -> ab + " " + c); 

Answer 7

public static <T> CompletableFuture<T> fromFuture(Future<T> f) {
    return CompletableFuture.completedFuture(null).thenCompose(avoid -> {
        try {
            return CompletableFuture.completedFuture(f.get());
        } catch (InterruptedException e) {
            return CompletableFuture.failedFuture(e);
        } catch (ExecutionException e) {
            return CompletableFuture.failedFuture(e.getCause());
        }
    });
}