我什么时候应该在 ExecutorService 上使用 CompletionService？

Question

我刚刚在 这篇博 文中找到了 CompletionService。但是，这并没有真正展示 CompletionService 相对于标准 ExecutorService 的优势。可以使用任何一种编写相同的代码。那么，什么时候 CompletionService 有用呢？

您能否提供一个简短的代码示例以使其清晰明了？例如，此代码示例仅显示不需要 CompletionService 的位置（= 相当于 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();
    }

原文由 ripper234 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

使用 ExecutorService ，提交要运行的任务后，您需要手动编码以有效地获取完成任务的结果。

使用 CompletionService ，这几乎是自动化的。您提供的代码中的区别不是很明显，因为您只提交了一项任务。但是，假设您有一个要提交的任务列表。在下面的示例中，多个任务被提交给 CompletionService。然后，它不会试图找出哪个任务已完成（以获取结果），它只是要求 CompletionService 实例在结果可用时返回结果。

 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");
                }
            }
        }
    }

原文由 Bhaskar 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

省略许多细节：

• ExecutorService = 传入队列 + 工作线程
• CompletionService = 传入队列 + 工作线程 + 输出队列

原文由 Alex Miller 发布，翻译遵循 CC BY-SA 3.0 许可协议