java CompletionService ExecutorCompletionSerivce

我们来想一个问题：

如果向Executor提交了一组计算任务，并且希望在计算完成后获得结果，那么我们可以保留与每个任务关联的Future，然后反复使用get方法，从而通过轮询来拿到返回结果，但是这样有些繁琐。 

而ExecutorCompletionService可以简化这个操作。这就是他的作用，可以对比一下两个代码。

废话不说，上代码。

package com.citi.test.mutiplethread.demo0511;

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;

public class TestCompletionService {
    public static void main(String[] args) {
        ExecutorService service=Executors.newFixedThreadPool(10);
        List<Future<Integer>> list=new ArrayList<Future<Integer>>(); //跟CompletionService比，多余的步骤
        for(int i=0;i<9;i++){
            final int index=i;
            Future<Integer> submit = service.submit(new Callable<Integer>() { //这句话前面部分也是多余的步骤
                @Override
                public Integer call() throws Exception {
//                    int a=1/0;
                    return index;
                }
            });
            list.add(submit);
        }
        for(Future<Integer> temp: list){
            try {
                Integer integer = temp.get();
                System.out.println(Thread.currentThread().getName()+" "+integer);
            } catch (InterruptedException | ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        service.shutdown();
    }
}

可以看到上面的方法也是可以的，但是有些繁琐，java提供了一种更好的方法：完成服务(CompletionSerivce)

不多说，上代码，

package com.citi.test.mutiplethread.demo0511;

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.Future;

public class TestCompletionService2 {
    public static void main(String[] args) {
        ExecutorService executor=Executors.newCachedThreadPool();
        CompletionService<Integer> completionService= new ExecutorCompletionService<Integer>(executor);
        for(int i=0;i<10;i++){
            final int index=i;
            completionService.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    return index;
                }
            });
        }
        for(int i=0;i<10;i++){
            try {
                Future<Integer> take = completionService.take(); //可以看到我们可以直接用completionService 来拿到Future对象。非常简便。
                Integer integer = take.get();
                System.out.println(Thread.currentThread().getName()+" "+integer);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (ExecutionException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        executor.shutdown();
    }
}

可以看到，我们不用像原来那样，

在向ExecutorService中提交任务之后，都去拿Future对象，

而是在提交完所有任务后，直接通过completionService.take来拿future对象了。

说一下原理，通过表象可以看出

CompletionService将Executor和BlockingQueue的功能融合在一起，你可以将callable任务提交给它执行，然后使用类似队列的操作的take和poll等方法来获得已完成的结果。

ExecutorCompletionSerivce实现了CompletionService，将计算部分委托给一个Executor。

通过源码可以看到，

CompletionService 这是个接口，定义了一些规范。

ExecutorCompletionSerivce是CompletionService的实现类，在构造函数中创建了一个BlockingQueue来保存计算完成的结果。

当提交某个任务时，该任务首先被包装为一个QueueingFuture，这是FutureTask的一个子类，然后改写子类的done方法，并将结果放到这个BlockingQueue中，之后想拿结果时，可以直接用BlockingQueue的take和poll方法来获取结果，这里而且是阻塞的。

成员变量和内部类

//成员变量。
private final Executor executor; 
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;

//继承自FutureTask，重写done方法，来实现在任务完成后，把结果加入到BlockingQueue中
/**
 * FutureTask extension to enqueue upon completion
 */
private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) {
        super(task, null);
        this.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future<V> task;
}

构造方法有两个

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

//可以传入想用的阻塞队列
public ExecutorCompletionService(Executor executor,
                                 BlockingQueue<Future<V>> completionQueue) {
    if (executor == null || completionQueue == null)
        throw new NullPointerException();
    this.executor = executor;
    this.aes = (executor instanceof AbstractExecutorService) ?
        (AbstractExecutorService) executor : null;
    this.completionQueue = completionQueue;
}

submit方法具体实现

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    //将callable提交给ExecutorCompletionSerivce之后，任务会被包装成QueueingFuture,而QueueingFuture继承自FutureTask，只不过重写了done方法，重写之后的done方法，可以看到是把包装之后的task放入到了阻塞队列中，这样在提交完所有任务之后，我们可以直接调用ExecutorCompletion.take方法来获得结果了
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture(f));
    return f;
}

private RunnableFuture<V> newTaskFor(Callable<V> task) {
    if (aes == null)
        return new FutureTask<V>(task);
    else
        return aes.newTaskFor(task);
}

private class QueueingFuture extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task) {
        super(task, null);
        this.task = task;
    }
    protected void done() { completionQueue.add(task); }
    private final Future<V> task;
}