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 CompletableFuture s, которые находятся в игре:

1. тот, который вернулся runAsync()
2. тот, который вернулся myMethod()
3. тот, который вернулся thenCompose()

У вас также есть 4 задачи, которые нужно выполнить:

1. тот, который передан runAsync(), будет выполнен на данном исполнителе и обработает будущее 1;
2. тот, который вызывает myMethod() из thenCompose() для создания будущего 2, может быть запущен на любом исполнителе, используйте thenComposeAsync(), чтобы явно выбрать один;
3. тот, который завершит будущее 2, возвращаемый myMethod() - это будет контролироваться внутри самого myMethod();
4. тот, который завершит будущее 3, возвращенное thenCompose() - это обрабатывается внутренне и будет зависеть от порядка выполнения (например, если 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();