Quand dois-je utiliser un CompletionService sur un ExecutorService?

Question

Je viens de trouver CompletionService dans ce blog. Cependant, ce n'est pas vraiment présente les avantages de CompletionService sur une norme ExecutorService. Le même code peut être écrit avec les deux. Donc, quand on est un CompletionService utile?

Pouvez-vous donner un court exemple de code pour faire ce cristal clair? Par exemple, cet exemple de code montre où un CompletionService n'est pas nécessaire (=équivalent à ExecutorService)

    ExecutorService taskExecutor = Executors.newCachedThreadPool();
    //        CompletionService<Long> taskCompletionService =
    //                new ExecutorCompletionService<Long>(taskExecutor);
    Callable<Long> callable = new Callable<Long>() {
        @Override
        public Long call() throws Exception {
            return 1L;
        }
    };

    Future<Long> future = // taskCompletionService.submit(callable);
        taskExecutor.submit(callable);

    while (!future.isDone()) {
        // Do some work...
        System.out.println("Working on something...");
    }
    try {
        System.out.println(future.get());
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (ExecutionException e) {
        e.printStackTrace();
    }

Answer 1

En omettant de nombreux détails:

• ExecutorService = file d'attente entrante + threads de travail
• CompletionService = file d'attente entrante + threads + file d'attente de sortie

Answer 2

Avec ExecutorService , une fois que vous avez soumis les tâches à exécuter , vous devez coder manuellement efficacement pour obtenir les résultats des tâches accomplies. Avec CompletionService , c'est à peu près automatisé. La différence n'est pas très évidente dans le code que vous avez présenté, parce que vous soumettez à une seule tâche. Cependant , imaginez que vous avez une liste de tâches à être soumis. Dans l'exemple ci-dessous , plusieurs tâches sont soumis à la CompletionService. Alors , au lieu d'essayer de savoir quelle tâche est terminée ( pour obtenir les résultats ) , il demande juste le CompletionService exemple pour renvoyer le résultat à mesure qu'ils deviennent disponibles.

public class CompletionServiceTest {

        class CalcResult {
             long result ;

             CalcResult(long l){
                 result = l;
             }
        }

        class CallableTask implements Callable<CalcResult> {
            String taskName ;
            long  input1 ;
            int input2 ;

            CallableTask(String name , long v1 , int v2 ){
                taskName = name;
                input1 = v1;
                input2 = v2 ;
            }

            public CalcResult call() throws Exception {
                System.out.println(" Task " + taskName + " Started -----");
                for(int i=0;i<input2 ;i++){
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        System.out.println(" Task " + taskName + " Interrupted !! ");
                        e.printStackTrace();
                    }
                    input1 += i;
                }
                System.out.println(" Task " + taskName + " Completed @@@@@@");
                return new CalcResult(input1) ;
            }

        }

        public void test(){
            ExecutorService taskExecutor = Executors.newFixedThreadPool(3);
                CompletionService<CalcResult> taskCompletionService =
                   new ExecutorCompletionService<CalcResult>(taskExecutor);

            int submittedTasks = 5;
            for(int i=0;i< submittedTasks;i++){
                taskCompletionService.submit(new CallableTask(
                        String.valueOf(i), 
                        (i * 10), 
                        ((i * 10) + 10  )
                        ));
               System.out.println("Task " + String.valueOf(i) + "subitted");
            }
            for(int tasksHandled=0;tasksHandled<submittedTasks;tasksHandled++){
                try {
                    System.out.println("trying to take from Completion service");
                    Future<CalcResult> result = taskCompletionService.take();
                    System.out.println("result for a task availble in queue.Trying to get()"  );
                    // above call blocks till atleast one task is completed and results availble for it
                    // but we dont have to worry which one

                    // process the result here by doing result.get()
                    CalcResult l = result.get();
                    System.out.println("Task " + String.valueOf(tasksHandled) + "Completed - results obtained : " + String.valueOf(l.result));

                } catch (InterruptedException e) {
                    // Something went wrong with a task submitted
                    System.out.println("Error Interrupted exception");
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    // Something went wrong with the result
                    e.printStackTrace();
                    System.out.println("Error get() threw exception");
                }
            }
        }
    }

Answer 3

Fondamentalement, vous utilisez un CompletionService si vous voulez exécuter plusieurs tâches en parallèle, et ensuite travailler avec eux dans leurs achèvement de la commande. Donc, si j'execute 5 emplois, l' CompletionService me donnera la première que cela se termine. L'exemple où il n'existe qu'une seule tâche ne confère aucune valeur supplémentaire sur un Executor outre la possibilité de soumettre un Callable.

Answer 4

Je pense que la javadoc meilleures réponses à la question de savoir quand l' CompletionService est utile dans un façon une ExecutorService ne l'est pas.

Un service qui découple la production de nouvelles tâches asynchrones à partir de la consommation des résultats des tâches accomplies.

Fondamentalement, cette interface permet à un programme de disposer de producteurs qui créer et soumettre des tâches (et même examiner les résultats de ces moyens) sans savoir à propos de tous les autres consommateurs, les résultats de ces tâches. Pendant ce temps, les consommateurs sont conscients de l' CompletionService peut poll ou take résultats sans être au courant des producteurs de soumettre les tâches.

Pour l'enregistrement, et j'ai peut-être tort, car il est un peu tard, mais je suis assez certain que le code d'exemple dans ce billet de blog provoque une fuite de mémoire. Sans un consommateur actif prenant les résultats de l' ExecutorCompletionServices'intérieur de la file d'attente, je ne suis pas sûr de savoir comment le blogueur attend à ce que la file d'attente à l'égout.

Answer 5

Tout d'abord, si nous ne voulez pas gaspiller de temps processeur, nous n'utiliserons pas

while (!future.isDone()) {
        // Do some work...
}

Nous devons utiliser

service.shutdown();
service.awaitTermination(14, TimeUnit.DAYS);

La mauvaise chose à propos de ce code est qu'il s'arrête ExecutorService. Si nous voulons continuer à travailler avec elle (c'est à dire que nous avons une certaine recursicve la création de la tâche), nous avons deux solutions: invokeAll ou ExecutorService.

invokeAll va attendre jusqu'à ce que tous les tâches à remplir. ExecutorService nous donne la capacité de prendre ou de résultats d'un sondage, un par un.

Et, finily, récursive exemple:

ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUMBER);
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<String>(executorService);

while (Tasks.size() > 0) {
    for (final Task task : Tasks) {
        completionService.submit(new Callable<String>() {   
            @Override
            public String call() throws Exception {
                return DoTask(task);
            }
        });
    } 

    try {                   
        int taskNum = Tasks.size();
        Tasks.clear();
        for (int i = 0; i < taskNum; ++i) {
            Result result = completionService.take().get();
            if (result != null)
                Tasks.add(result.toTask());
        }           
    } catch (InterruptedException e) {
    //  error :(
    } catch (ExecutionException e) {
    //  error :(
    }
}