Jdk1.6 JUC源码解析(21)-ExecutorCompletionService

Jdk1.6 JUC源码解析(21)-ExecutorCompletionService

作者:大飞

 

功能简介:

  • ExecutorCompletionService用于执行一批任务,然后按照任务执行完成的顺序来获取任务结果。你甚至可以在获取到了若干个执行结果后,把其他的任务取消掉(ThreadPoolExecutor中的invokeAny就是通过这货实现的)。比如这样的场景:你的业务需要调用10个接口来获取一些信息,业务规定只需要其中任意2个接口的信息,那么就可以使用ExecutorCompletionService,获取前两个成功完成的任务结果,然后将其他的任务取消。

源码分析:

  • ExecutorCompletionService实现了CompletionService接口,先看下这个接口:
public interface CompletionService<V> {
    /**
     * 提交一个有返回值的任务。
     * 一旦这个任务完成,就能通过taken或者poll方法来获取任务结果。
     */
    Future<V> submit(Callable<V> task);
    /**
     * 提交一个Runnable和一个返回值。
     * 一旦这个任务完成,就能通过taken或者poll方法来获取这个返回值。
     */
    Future<V> submit(Runnable task, V result);
    /**
     * 获取并移除下一个完成的任务,如果当前没有任务完成,阻塞等待。
     */
    Future<V> take() throws InterruptedException;
    /**
     * 获取并移除下一个完成的任务,如果当前没有任务完成,返回null。
     */
    Future<V> poll();
    /**
     * 获取并移除下一个完成的任务,如果当前没有任务完成,阻塞等待,
     * 如果在超时前仍然没有任务完成,返回null。
     */
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

 

 

  • 看下ExecutorCompletionService的代码,首先看下内部结构:

public class ExecutorCompletionService<V> implements CompletionService<V> {
    private final Executor executor;
    private final AbstractExecutorService aes;
    private final BlockingQueue<Future<V>> completionQueue;
    /**
     * 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;
    }

       通过代码可见,内部有用来实际执行任务的executor和用来存放完成任务的阻塞队列(注意到内部的QueueingFuture覆盖了FutureTask的钩子方法done,在任务完成后会将其放到完成队列里面)。

       看下构造方法:

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

         构造方法中,当传入的executor属于AbstractExecutorService类型,会将这个executor同时赋给aes,主要用于内部包装异步任务: 

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

   

  • 最后看下功能方法:

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

       方法很简单,提交到ExecutorCompletionService的任务会在内部被包装成QueueingFuture,并由内部的executor来执行这个任务,当任务执行完成后,会被加入到内部的队列里面,外部程序就可以通过take或者poll方法来获取完成的任务了。

           ExecutorCompletionService的代码解析完毕!          参见:
Jdk1.6 JUC源码解析(17)-ThreadPoolExecutor  

    原文作者:JUC
    原文地址: https://blog.csdn.net/iteye_11160/article/details/82642675
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞