Use of completionservice in Java

Use of completionservice in Java

In the previous article, we talked about executorservice. Through executorservice, we can submit tasks one by one, return to future, and then call future Get method to return the execution result of the task.

Although this method is effective, it is troublesome to save each returned future value. Fortunately, executorservice provides an invokeall method to save all future values. Let's see a specific implementation:

   public void useExecutorService() throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);

        Callable<String> callableTask = () -> {
            TimeUnit.MILLISECONDS.sleep(300);
            return "Task's execution";
        };

        List<Callable<String>> callableTasks = new ArrayList<>();
        callableTasks.add(callableTask);
        callableTasks.add(callableTask);
        callableTasks.add(callableTask);

        List<Future<String>> futures = executor.invokeAll(callableTasks);

        executor.shutdown();

    }

In the above example, we defined three tasks by calling executor Invokeall (callabletasks) returns a list < future >, so we can get all the returned values.

In addition to the invokeall method above, we will introduce a completionservice interface today.

Completionservice is actually a combination of executorservice and BlockingQueue. Executorservice is used to submit tasks, and BlockingQueue is used to save execution results encapsulated into future. Get the future value by calling the take and poll methods.

Completionservice is an interface. Let's take a look at its specific implementation executorcompletionservice:

    public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }

Executorcompletionservice receives an executor as a parameter.

Let's take a look at the above example. What if you rewrite it with executorcompletionservice:

   public void useCompletionService() throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CompletionService<String> completionService=new ExecutorCompletionService<String>(executor);
        Callable<String> callableTask = () -> {
            TimeUnit.MILLISECONDS.sleep(300);
            return "Task's execution";
        };
        for(int i=0; i< 5; i ++){
            completionService.submit(callableTask);
        }

        for(int i=0; i<5; i++){
            Future<String> result=completionService.take();
            System.out.println(result.get());
        }
    }

The above example uses completionservice Submit to submit the task through completionservice Take() to get the result value.

In fact, completionservice also has a poll method. The difference between poll and take is that take will wait if it cannot get the value, while poll will return null.

Examples of this article can be referred to https://github.com/ddean2009/learn-java-concurrency/tree/master/CompletionService

For more information, please visit flybean's blog