java中CompletionService的使用

java中CompletionService的使用

之前的文章中我們講到了ExecutorService，通過ExecutorService我們可以提交一個個的task，並且返回Future，然后通過調用Future.get方法來返回任務的執行結果。

這種方式雖然有效，但是需要保存每個返回的Future值，還是比較麻煩的，幸好ExecutorService提供了一個invokeAll的方法，來保存所有的Future值，我們看一個具體的實現：

   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();

    }

上面的例子中，我們定義了3個task，通過調用executor.invokeAll(callableTasks)返回了一個 List<Future>，這樣我們就可以得到所有的返回值了。

除了上面的invokeAll方法外，我們今天要介紹一個CompletionService接口。

CompletionService實際上是ExecutorService和BlockingQueue的結合體，ExecutorService用來提交任務，而BlockingQueue用來保存封裝成Future的執行結果。通過調用take和poll的方法來獲取到Future值。

CompletionService是一個接口，我們看下它的一個具體實現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接收一個Executor作為參數。

我們看下上面的例子如果用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());
        }
    }

上面的例子通過completionService.submit來提交任務，通過completionService.take()來獲取結果值。

其實CompletionService還有一個poll的方法，poll和take的區別在於：take如果獲取不到值則會等待，而poll則會返回null。

本文的例子可以參考https://github.com/ddean2009/learn-java-concurrency/tree/master/CompletionService

更多內容請訪問 flydean的博客