ExecutorCompletionService源碼解析

主要來看看FutureTask的done方法的運用。

public class ExecutorCompletionService<V> implements CompletionService<V> {
	//執行任務的線程池
    private final Executor executor;
    //用于調用AbstractExecutorService的newTaskFor方法，來執行個體化一個實作了RunnableFuture接口的對象
    //如果executor繼承了AbstractExecutorService ，則直接調用executor的newTaskFor方法
    //否則直接建立一個FutureTask對象
    private final AbstractExecutorService aes;
    //任務完成後放入該阻塞隊列中
    private final BlockingQueue<Future<V>> completionQueue;

     //用于放入執行完成的任務
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        //重寫了FutureTask的done方法，任務完成後，将任務放入阻塞隊列中
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }
	
	//将傳入的Callable包裝為RunnableFuture
    private RunnableFuture<V> newTaskFor(Callable<V> task) {
        if (aes == null)
            return new FutureTask<V>(task);
        else
            return aes.newTaskFor(task);
    }
	//将傳入的Callable包裝為RunnableFuture
    private RunnableFuture<V> newTaskFor(Runnable task, V result) {
        if (aes == null)
            return new FutureTask<V>(task, result);
        else
            return aes.newTaskFor(task, result);
    }

 	//completionQueue預設為LinkedBlockingQueue
    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;
    }
	//送出任務，任務被包裝為QueueingFuture對象，主要重寫FutureTask的done方法，使得任務執行完畢後被執行任務的線程放入到阻塞隊列中
    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;
    }

    public Future<V> submit(Runnable task, V result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task, result);
        executor.execute(new QueueingFuture(f));
        return f;
    }
	
	//從阻塞隊列中擷取任務
    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }

    public Future<V> poll() {
        return completionQueue.poll();
    }

    public Future<V> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
        return completionQueue.poll(timeout, unit);
    }

}