博客

CompletionService异步获取并行任务执行结果

源码分析

首先看下构造方法

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

构造方法用来初始化一个阻塞对垒， 用来存储已完成的task 任务

然后看下submit 方法

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    //将我们的callable任务包装成QueueingFuture
    executor.execute(new QueueingFuture(f));
    return f;
}

submit 方法可以看出 callable 被包装成了QueueFuture ，而QueueingFutrue 继承  FutureTask ，所以最终执行了FutureTask 中的run 方法

public void run() {
 //判断执行状态，保证callable任务只被运行一次
    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 {
            //这里回调我们创建的callable对象中的call方法
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran)
            //处理执行结果
                set(result);
        }
    } finally {
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

在run 方法中， 最终回调自定义的callable 中的call 方法， 执行结束后，通过 set 方法执行结果

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    //设置执行结果v，并标记线程执行状态为：NORMAL
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); 
        //完成执行，将执行结果添加到队列
        finishCompletion();
    }
}

我们在看下 finishCompletion() 方法， 在该方法中我们找到 done()

protected void done() { completionQueue.add(task); }

该方法将执行结束的task 添加到队列中， 只要队列中有元素， 我们调用 take() 方法就可以获取执行的结果，异步非阻塞获取执行结果的实现原理，其实就是通过队列来实现的， FutureTask将执行结果放到队列中， 先进先出， 线程执行结束的顺序就是获取结果的顺序。