Exécuteurs Java : comment être notifié, sans blocage, de la fin d'une tâche ?

Question

Disons que j'ai une file d'attente pleine de tâches que je dois soumettre à un service exécuteur. Je veux qu'elles soient traitées une par une. La méthode la plus simple à laquelle je pense est la suivante :

1. Prendre une tâche dans la file d'attente
2. Soumettez-le à l'exécuteur testamentaire
3. Appeler .get sur le Future retourné et bloquer jusqu'à ce qu'un résultat soit disponible.
4. Prendre une autre tâche dans la file d'attente...

Cependant, j'essaie d'éviter complètement le blocage. Si j'ai 10 000 files d'attente de ce type, qui ont besoin que leurs tâches soient traitées une par une, je manquerai d'espace sur la pile parce que la plupart d'entre elles s'accrocheront à des threads bloqués.

Ce que je voudrais, c'est soumettre une tâche et fournir un rappel qui est appelé lorsque la tâche est terminée. Je vais utiliser cette notification de rappel comme un drapeau pour envoyer la tâche suivante. (functionaljava et jetlang utilisent apparemment de tels algorithmes non bloquants, mais je n'arrive pas à comprendre leur code)

Comment puis-je le faire en utilisant java.util.concurrent du JDK, sans écrire mon propre service d'exécution ?

(la file d'attente qui m'alimente en tâches peut elle-même se bloquer, mais c'est un problème qui sera abordé plus tard).

Answer 1

Définissez une interface de rappel pour recevoir les paramètres que vous souhaitez transmettre dans la notification d'achèvement. Puis invoquez-la à la fin de la tâche. Vous pouvez même écrire une enveloppe générale pour les tâches Runnable et les soumettre à l'interface ExecutorService .

class CallbackTask implements Runnable {

  private final Runnable task;

  private final Callback callback;

  CallbackTask(Runnable task, Callback callback) {
    this.task = task;
    this.callback = callback;
  }

  public void run() {
    task.run();
    callback.complete();
  }

}

Answer 2

En Java 8, vous pouvez utiliser CompletableFuture . Voici un exemple dans mon code où je l'utilise pour récupérer les utilisateurs de mon service utilisateur, les associer à mes objets de vue, puis mettre à jour ma vue ou afficher un dialogue d'erreur (il s'agit d'une application GUI) :

    CompletableFuture.supplyAsync(
            userService::listUsers
    ).thenApply(
            this::mapUsersToUserViews
    ).thenAccept(
            this::updateView
    ).exceptionally(
            throwable -> { showErrorDialogFor(throwable); return null; }
    );

Il s'exécute de manière asynchrone. J'utilise deux méthodes privées : listUsers et updateView .

Answer 3

Utilisez L'API du futur écoutable de Guava et ajouter un callback. Cf. le site web :

ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
ListenableFuture<Explosion> explosion = service.submit(new Callable<Explosion>() {
  public Explosion call() {
    return pushBigRedButton();
  }
});
Futures.addCallback(explosion, new FutureCallback<Explosion>() {
  // we want this handler to run immediately after we push the big red button!
  public void onSuccess(Explosion explosion) {
    walkAwayFrom(explosion);
  }
  public void onFailure(Throwable thrown) {
    battleArchNemesis(); // escaped the explosion!
  }
});

Answer 4

Vous pourriez étendre la classe FutureTask, et surcharger la méthode done(), puis ajouter l'objet FutureTask à l'ExecutorService, de sorte que la méthode done() sera invoquée lorsque la FutureTask sera terminée immédiatement.

Answer 5

ThreadPoolExecutor a également beforeExecute et afterExecute des méthodes d'accrochage que vous pouvez remplacer et utiliser. Voici la description de ThreadPoolExecutor 's javadocs .

Méthodes d'accrochage

Cette classe fournit des fonctions protégées et redéfinissables beforeExecute(java.lang.Thread, java.lang.Runnable) et afterExecute(java.lang.Runnable, java.lang.Throwable) qui sont appelées avant et après l'exécution de chaque tâche. Elles peuvent être utilisées pour manipuler l'environnement d'exécution ; par exemple, réinitialiser les ThreadLocals, recueillir des statistiques ou ajouter des entrées de journal. De plus, les méthodes terminé() peut être surchargé pour effectuer tout traitement spécial qui doit être fait une fois que l'exécuteur est complètement terminé. Si les méthodes de hook ou de callback lèvent des exceptions, les fils de travail internes peuvent à leur tour échouer et se terminer brusquement.