Java CompletableFuture 分配执行器

Question

我对在 CompletableFuture 中定义执行者感到困惑。我不确定如何告诉 CompletableFuture 在该特定执行程序中运行它。提前致谢。

//Suppose I have an executor
ExecutorService myExecutor=Executors.newFixedThreadPool(2);

//If I create a future like this
CompletableFuture.runAsync(() -> {
      //Do something
}, myExecutor); // I can put the executor here and say the future to this executor

//But I do not know where to put executor if I create my future in method style like this

private final CompletableFuture<Void> myMethod(String something) {
  //Do something
    return null;
}

//and use it like this  
.thenCompose(this::myMethod); //How can I specify the executor in this case?

Answer 1

在您的示例中，您有 3 个 CompletableFutures 在起作用：

1. runAsync()返回的那个
2. myMethod()返回的那个
3. thenCompose()返回的那个

您还有 4 个任务需要运行：

1. 传递给runAsync() 的那个将在给定的执行器上执行并处理未来1；
2. 从thenCompose()调用myMethod()来创建future 2的那个可以在任何executor上运行，使用thenComposeAsync()明确选择一个；
3. 将完成由myMethod() 返回的future 2 - 这将在myMethod() 自身内部进行控制；
4. 将完成由 thenCompose() 返回的未来 3 的那个——这是在内部处理的，取决于执行顺序（例如，如果 myMethod() 返回一个已经完成的未来，它也会完成前者）。

如您所见，涉及多个任务和执行器，但您始终可以使用*Async() 变体控制在独立阶段中使用的执行器。您无法真正控制它的唯一情况是第 4 种情况，但只要相关阶段也使用 *Async() 变体，这是一种廉价的操作。

Answer 2

你可以这样做：

ExecutorService es = Executors.newFixedThreadPool(4);
List<Runnable> tasks = getTasks();
CompletableFuture<?>[] futures = tasks.stream()
                               .map(task -> CompletableFuture.runAsync(task, es))
                               .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();    
es.shutdown();