CompletionService详解

当我们有批量执行多个异步任务的业务，并且需要调用get()获取执行结果的时候，我们写的代码一般如下：

Future f1 = excutor.submit(c1);
f1.get();
Future f2 = excutor.submit(c2);
f2.get();

f1.get()在获取成功之前会被阻塞，会阻塞c2的执行，严重降低了效率。

这时我们可以通过CompletionService来解决这一问题！

CompletionService结构

image.png

CompletionSerive接口有一个实现类ExecutorCompletionService，结构如下

image.png

实例

BlockingQueue q = new LinkedBlockingQueue();
ThreadPoolExecutor executor = new ThreadPoolExecutor(4,6,3, TimeUnit.SECONDS,q);
CompletionService<Integer> cs = new ExecutorCompletionService<>(executor);

cs.submit(new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        Thread.sleep(2000);
        return 1;
    }
});
cs.submit(new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        Thread.sleep(2000);
        return 1;
    }
});


for (int i = 0;i < 3;i++){
    Integer f = cs.take().get();
    System.out.println(f);
}

源码解析

public ExecutorCompletionService(Executor executor,
                                 BlockingQueue<Future<V>> completionQueue) {
    if (executor == null || completionQueue == null)
        throw new NullPointerException();
    //线程池
    this.executor = executor;
    //AbstractExecutorService
    this.aes = (executor instanceof AbstractExecutorService) ?
        (AbstractExecutorService) executor : null;
    //保存Future的队列
    this.completionQueue = completionQueue;
}

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    //包装成FutureTask
    RunnableFuture<V> f = newTaskFor(task);
    //会调用FutureTask的run()
    executor.execute(new QueueingFuture(f));
    return f;
}

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


public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    //调用call()方法，result为执行的结果
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                //这里这里这里
                    set(result);
            }
        } finally {

            runner = null;
            
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
}

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        //将FutureTask的outcome设置result
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
        //这里这来这里
        finishCompletion();
    }
}

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }
    //将task加入到completionQueue
    done();

    callable = null;        // to reduce footprint
}

总结
从源码解析看，ExecutorCompletionService中内置了一个Future的completionQueue，在任务调用完成后，会将submit返回的future放入到completionQueue。用户可以通过completionQueue.take()得到future然后调用future.get()来获取任务执行结果。