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