Java使用Executor執行Callable任務時的幾種方法

多線程在須要返回值時，咱們知道須要用到Callable和Future。Callable的cell方法能夠返回一個值而且可拋出異常，是對Runnable的很好的補充；Future表示了一個任務的週期，它提供了判斷任務狀態、獲取任務結果和取消任務等方法 。
下面演示三種使用Executor執行Callable任務的方法。

/**
     * 測試任務，返回任務的序號
     */
    public static class TestTask implements Callable<Integer>{
        int index;
        public TestTask(int index) {
            this.index = index;
        }
        @Override
        public Integer call() throws Exception {
            return index;
        }
    }
	
     /**
     * 方法一：手動的保存任務的返回，這樣的好處是每一個任務對應的結果咱們很清楚
     */
    @Test
    public void ordinaryTest() throws ExecutionException, InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(10);
        List<Future<Integer>> futures = new ArrayList<>();
        for(int i = 0; i < 10; i++) {
            TestTask testTask = new TestTask(i);
            Future<Integer> future = es.submit(testTask);
            futures.add(future);
        }
        es.shutdown();
        for(int i = 0; i < 10; i++) {
            System.out.println("index:" + i + ",future:"+ futures.get(i).get());
        }
    }

    /**
     * 方法二：使用ExecutorCompletionService
     * ExecutorCompletionService中使用阻塞隊列保存各任務的返回結果，返回是無序的，即誰先執行完成（異常、中斷），誰先入隊。
     * 當咱們不關心結果的順序，或者須要一個任務完成時就取消其餘任務的狀況下，它是很是的方便的
     */
    @Test
    public void completionServiceTest() throws ExecutionException, InterruptedException {
        ExecutorService es = Executors.newFixedThreadPool(10);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(es);
        for(int i = 0; i < 10; i++) {
            TestTask testTask = new TestTask(i);
            completionService.submit(testTask);
        }
        es.shutdown();
        for(int i = 0; i < 10; i++) {
            Future<Integer> future = completionService.take();
            System.out.println("index:" + i + ",future:"+ future.get());
        }
    }

    /**
     * 方法三：ExecutorService的invokeAll方法
     * invokeAll方法入參爲一組任務，返回一組Future，這兩個集合是有相同結構的，
     * 即它是按照入參的任務集合中迭代器的順序將全部的Future添加到返回的集合中，從而任務和Future在它們各自的集合中有着一樣的順序。
     * 當咱們須要任務和結果的對應關係時，使用invokeAll方法來代替第一種方法
     */
    @Test
    public void invokeAllTest() throws InterruptedException, ExecutionException {
        ExecutorService es = Executors.newFixedThreadPool(10);
        List<TestTask> tasks = new ArrayList<>();
        for (int i = 0; i < 10; i++){
            tasks.add(new TestTask(i));
        }
        List<Future<Integer>> futures = es.invokeAll(tasks);
        es.shutdown();
        for (int i = 0; i < futures.size(); i++){
            System.out.println("index:" + i + ",future:"+ futures.get(i).get());
        }
    }

下面看一下ExecutorCompletionService的原理：
ExecutorCompletionService是將Executor和BlockingQueue的功能融合在一塊兒，可將Callbale任務提交給它來執行，而後咱們就能夠像隊列同樣使用take或poll來獲得已經完成的任務結果。下面是源碼分析：

/**
     *ExecutorCompletionService包含三個成員變量，最主要的是completionQueue，它的類型阻塞隊列
     */

    private final Executor executor;
    private final AbstractExecutorService aes;
    private final BlockingQueue<Future<V>> completionQueue;

    /**
     * 構造方法，須要咱們傳入一個Executor
     */
    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>>();
    }

    /**
     * 提交任務的方法，其中的RunnableFuture爲一個內部類，繼承自FutureTask
     */
    public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture(f));
        return f;
    }
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }

        /**
         * 這是QueueingFuture存在的主要緣由，當任務執行完成後，將任務結果裝入隊列中
         */
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

    /**
     * 從隊列中獲取返回值
     */
    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }