CompletionService小記

CompletionService小記

在使用ExecutorService時，經過sumit執行一個Callable的時候，會當即返回一個異步任務結果，而後經過get獲取異步任務結果的時候會阻塞，以下面這種狀況，代碼以下，

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * Created by xinxingegeya on 16/3/22.
 */
public class ExecutorServiceTest {

    public static void main(String args[])
            throws ExecutionException, InterruptedException {

        ExecutorService executor = Executors.newCachedThreadPool();

        List<Future<Integer>> futureList = new ArrayList<>();
        for (int i = 5; i > 0; i--) {
            final int taskID = i;
            //CompletionService.submit()
            Future<Integer> res = executor.submit(new Callable<Integer>() {
                public Integer call() throws Exception {
                    Thread.sleep(taskID * 1000);
                    return taskID;
                }
            });
            futureList.add(res);
        }

        for (Future<Integer> future : futureList) {
            int result = future.get();
            System.out.println(result);
        }

        executor.shutdown();

        while (executor.isTerminated()) {
            executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        }
    }
}

經過代碼能夠看到，在遍歷任務結果集的時候，獲取第一個任務結果，而後get會阻塞，會阻塞大概五秒的時間，而對於後面的任務，可能已經執行完成了，但就是阻塞在了第一Future上，這是不合理的，那麼如何改進，在這裏應該要使用CompletionService，以下代碼，

import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Created by xinxingegeya on 16/3/22.
 */
public class ExecutorCompletionServiceTest {

    public static void main(String args[]) throws InterruptedException {

        ExecutorService executor = Executors.newCachedThreadPool();

        CompletionService<Integer> cs = new ExecutorCompletionService<>(
            executor);
        for (int i = 5; i > 0; i--) {
            final int taskID = i;
            //CompletionService.submit()
            cs.submit(new Callable<Integer>() {
                public Integer call() throws Exception {
                    Thread.sleep(taskID * 1000);
                    return taskID;
                }
            });
        }
        // 可能作一些事情
        for (int i = 1; i < 5; i++) {
            try {
                int result = cs.take().get();//CompletionService.take()返回Future
                System.out.println(result);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        executor.shutdown();

        while (executor.isTerminated()) {
            executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        }

        System.out.println("done");
    }
}

使用CompletionService的好處就是優先獲取已經執行完成的任務結果，而不會一味的等待尚未執行完成的任務。

CompletionService的實現機制就是利用BlockingQueue，若是任務已經執行完成，會把Future放進阻塞隊列裏面，若是阻塞隊列中有已經完成的任務，那麼就取出來（take）獲取異步任務的結果。

==============END==============