Welcher Executor wird beim Erstellen von Java CompleableFutures verwendet?

Question

Ich habe eine Methode für eine Repository-Klasse, die eine CompletableFuture zurückgibt. Der Code, der diese Futures vervollständigt, verwendet eine Bibliothek von Dritten, die blockiert. Ich beabsichtige, eine separate begrenzte Executor zu haben, die diese Repository-Klasse verwenden wird, um diese blockierenden Aufrufe zu machen. Hier

ein Beispiel:

public class PersonRepository { 
    private Executor executor = ... 
    public CompletableFuture<Void> create(Person person) {...} 
    public CompletableFuture<Boolean> delete(Person person) {...} 
} 

Der Rest meiner Anwendung werden diese Futures und tun einige andere Dinge mit den Ergebnissen zusammenstellen. Wenn diese anderen Funktionen, die an thenAccept, thenCompose, whenComplete usw. geliefert werden, möchte ich nicht, dass sie auf dem Executor für das Repository ausgeführt werden.

Ein weiteres Beispiel:

public CompletableFuture<?> replacePerson(Person person) { 
    final PersonRepository repo = getPersonRepository(); 
    return repo.delete(person) 
     .thenAccept(existed -> { 
      if (existed) System.out.println("Person deleted")); 
      else System.out.println("Person did not exist")); 
     }) 
     .thenCompose(unused -> { 
      System.out.println("Creating person"); 
      return repo.create(person); 
     }) 
     .whenComplete((unused, ex) -> { 
      if (ex != null) System.out.println("Creating person"); 
      repo.close(); 
     }); 
} 

javadoc Zustände:

Aktionen für abhängige Beendigungen von nicht-Async Verfahren geliefert wird, kann durch den Thread durchgeführt werden, die den aktuellen CompletableFuture abgeschlossen ist, oder durch jede andere Aufrufer einer Abschlussmethode.

Side Frage: Warum gibt es eine oder hier? In welchem ​​Fall gibt es einen anderen Aufrufer einer Abschlussmethode, die die aktuelle Zukunft nicht abschließt?

Hauptfrage: Wenn ich will alle println von einem anderen Executor als die vom Repository verwendet ein ausgeführt werden, welche Methoden muss ich async machen und die Vollstrecker von Hand zur Verfügung stellen?

Offensichtlich muss die thenAccept in thenAcceptAsync geändert werden, aber ich bin nicht sicher über diesen Punkt weiter.

Alternative Frage: Welcher Thread vervollständigt die zurückgegebene Zukunft von ?

Meine Vermutung ist, dass es sein wird, was auch immer Thread die Zukunft aus dem Funktionsargument ergänzt. Mit anderen Worten, ich müsste auch whenComplete zu whenCompleteAsync ändern.

Vielleicht bin ich über komplizierende Dinge, aber das fühlt sich an, als könnte es ziemlich schwierig werden. Ich muss sehr darauf achten, woher all diese Futures kommen. Wenn ich eine Zukunft zurückgebe, wie kann ich verhindern, dass Anrufer meinen Executor verwenden? Es fühlt sich an, als ob es die Kapselung bricht. Ich weiß, dass alle Transformationsfunktionen in Scala eine implizite ExecutionContext annehmen, die all diese Probleme zu lösen scheint.

Answer 1

Side-Frage: Wenn Sie das Intermediate CompletionStage einer Variablen zugewiesen und eine Methode darauf aufgerufen haben, würde es auf dem gleichen Thread ausgeführt werden.

Hauptfrage: Nur die erste, also ändern thenAccept zu thenAcceptAsync - alle folgenden werden ihre Schritte auf den Thread ausführen, der für die Annahme verwendet wird.

Alternative Frage: Der Thread, der die Zukunft von thenCompose abgeschlossen hat, ist derselbe, der für den Compose verwendet wurde.

Sie sollten sich die CompletionStages als Schritte vorstellen, die in schneller Abfolge auf demselben Thread ausgeführt werden (indem Sie einfach die Funktionen der Reihe nach anwenden), es sei denn, Sie möchten den Schritt in einem anderen Thread asynchron ausführen. Alle nächsten Schritte werden dann für diesen neuen Thread ausgeführt.

In Ihrem aktuellen Setup würde die Schritte wie folgt ausgeführt werden:

Thread-1: delete, accept, compose, complete 

Mit dem ersten akzeptieren async, wird es:

Thread-1: delete 
Thread-2: accept, compose, complete 

Was Ihre letzte Frage, etwa die gleichen Thread Wesen von Ihren Anrufern verwendet, wenn sie zusätzliche Schritte hinzufügen - ich glaube nicht, dass es viel gibt, was Sie tun können, abgesehen davon, dass Sie keine CompletableFuture zurückgeben, sondern eine normale Future.

Answer 2

Nur aus meinen empirischen Beobachtungen beim Herumspielen hängt der Thread, der diese nicht-asynchronen Methoden ausführt, davon ab, was zuerst passiert, thenCompose selbst oder die Aufgabe hinter der Future.

Wenn thenCompose zuerst ausgeführt wird (was in Ihrem Fall fast sicher ist), wird die Methode auf demselben Thread ausgeführt, der die Future Aufgabe ausführt.

Wenn die Aufgabe hinter Ihrem Future zuerst abgeschlossen wird, wird die Methode sofort auf den aufrufenden Thread (d. H. Kein Executor) ausgeführt.