ExecutorService, comment attendre que toutes les tâches se terminent

Question

Quelle est la façon la plus simple d'attendre que toutes les tâches d' ExecutorService à la fin? Ma tâche est principalement de calcul, donc je tiens juste à exécuter un grand nombre d'emplois - un sur chaque cœur. Maintenant ma configuration ressemble à ceci:

ExecutorService es = Executors.newFixedThreadPool(2);
for (DataTable singleTable : uniquePhrases)
{   
    es.execute(new ComputeDTask(singleTable));
}
try 
{
    es.wait();
} 
catch (InterruptedException e) 
{
    e.printStackTrace();
}

ComputeDTask implements runnable. Cela semble d'exécuter les tâches correctement, mais le code se bloque sur wait() avec IllegalMonitorStateException. C'est étrange, car j'ai joué un peu avec certains jouets exemples et il est apparu à travailler.

uniquePhrases contient plusieurs dizaines de milliers d'éléments. Dois-je utiliser une autre méthode? Je suis à la recherche de quelque chose d'aussi simple que possible

Answer 1

L'approche la plus simple est d'utiliser ExecutorService.invokeAll() qui fait ce que vous voulez dans un one-liner. Dans votre langage, vous aurez besoin de les modifier ou de les envelopper ComputeDTask à mettre en oeuvre Callable<>, qui peut vous donner un peu plus de souplesse. Probablement dans votre application, il y a une mise en œuvre concrète des Callable.call(), mais voici une façon à l'envelopper si vous n'utilisez pas Executors.callable().

ExecutorService es = Executors.newFixedThreadPool(2);
List<Callable<Object>> todo = new ArrayList<Callable<Object>>(singleTable.size());

for (DataTable singleTable: uniquePhrases) { 
    todo.add(Executors.callable(new ComputeDTask(singleTable))); 
}

List<Future<Object>> answers = es.invokeAll(todo);

Comme d'autres l'ont souligné, vous pouvez utiliser le délai d'expiration de la version d' invokeAll() le cas échéant. Dans cet exemple, answers va contenir un tas de Futures qui retournera les valeurs null (voir définition de l' Executors.callable(). Probablement ce que vous voulez faire, c'est une légère refactoring de sorte que vous pouvez obtenir une réponse, ou une référence à la sous-jacentes ComputeDTask, mais je ne peux pas dire à partir de votre exemple.

Si elle n'est pas claire, note qu' invokeAll() ne sera pas de retour jusqu'à ce que toutes les tâches sont terminées. (c'est à dire, tous les Futures dans votre answers collection sera en rapport .isDone() si demandé.) Cela évite tous les arrêts manuels, awaitTermination, etc... et permet de réutiliser ce ExecutorService proprement pour plusieurs cycles, si vous le souhaitez.

Il y a quelques questions connexes AINSI:

• Comment attendre de tous les threads pour finir

• Les valeurs de retour de threads java

• invokeAll() pas prêts à accepter une Collection<Callable<t>>

• Ai-je besoin de synchroniser?

Aucune de ces sont strictement sur le point de votre question, mais ils fournissent un peu de couleur à propos de la façon dont les gens pensent Executor/ExecutorService doit être utilisé.

Answer 2

Si vous voulez attendre pour toutes les tâches à réaliser, l'utilisation de l' shutdown méthode de wait. Ensuite, il suivra avec awaitTermination.

Aussi, vous pouvez utiliser Runtime.availableProcessors pour obtenir le nombre de threads matériels de sorte que vous pouvez initialiser votre pool de threads correctement.

Answer 3

Si l'attente pour toutes les tâches dans l' ExecutorService à la finition n'est pas précisément votre objectif, mais plutôt d'attente jusqu'à ce qu'un lot spécifique de tâches est terminée, vous pouvez utiliser un CompletionService — plus précisément, ExecutorCompletionService.

L'idée est de créer un ExecutorCompletionService d'emballage de votre Executor, soumettre certains connu nombre de tâches par l'intermédiaire de l' CompletionService, puis de dessiner que le même nombre de résultats à partir de l' achèvement de la file d'attente en utilisant soit take() (qui bloque) ou poll() (qui ne fonctionne pas). Une fois que vous avez tiré tous les résultats attendus correspondant aux tâches que vous avez envoyé, vous savez qu'ils sont tous fait.

Permettez-moi de préciser ce une fois de plus, parce qu'il n'est pas évident à partir de l'interface: Vous devez savoir comment beaucoup de choses que vous mettez dans l' CompletionService afin de savoir comment beaucoup de choses pour essayer de faire ressortir. C'est important, surtout avec l' take() méthode: l'appeler une fois de trop, et il permet de bloquer votre thread appelant jusqu'à ce qu'un autre thread soumet un autre emploi de la même CompletionService.

Il y a quelques exemples montrant comment utiliser CompletionService dans le livre de Java de la Simultanéité dans la Pratique.

Answer 4

Si vous voulez attendre que le service exécutant ait fini d'exécuter, appelez shutdown() , puis waitTermination (unités, unitType) , par exemple awaitTermination(1, MINUTE) . L'executorService ne bloque pas sur son propre moniteur, vous ne pouvez donc pas utiliser wait etc.

Answer 5

Vous pourriez attendre d'emplois pour finir sur un certain intervalle:

int maxSecondsPerComputeDTask = 20;
try
{
    while (!es.awaitTermination(uniquePhrases.size() * maxSecondsPerComputeDTask, TimeUnit.SECONDS))
    {
        // consider giving up with a 'break' statement under certain conditions
    }
}
catch (InterruptedException e)
{
    throw new RuntimeException(e);    
}

Ou vous pourriez utiliser ExecutorService.soumettre(Runnable) et de recueillir l' Avenir des objets qu'il renvoie et appeler get() sur chacun à son tour, attendre pour eux à la fin.

ExecutorService es = Executors.newFixedThreadPool(2);
Collection<Future<?>> futures = new LinkedList<<Future<?>>();
for (DataTable singleTable : uniquePhrases)
{
    futures.add(es.submit(new ComputeDTask(singleTable)));
}
for (Future<?> future : futures)
{
   try
   {
       future.get();
   }
   catch (InterruptedException e)
   {
       throw new RuntimeException(e);
   }
   catch (ExecutionException e)
   {
       throw new RuntimeException(e);
   }
}

InterruptedException est extrêmement important de traiter correctement. C'est ce qui vous permet de vous ou d'autres utilisateurs de votre bibliothèque de mettre fin à un long processus en toute sécurité.