Java多线程 CompletionService

1 completionservice介绍

completionservice用于提交一组callable任务，其take方法返回已完成的一个callable任务对应的future对象。
如果你向executor提交了一个批处理任务，并且希望在它们完成后获得结果。为此你可以将每个任务的future保存进一个集合，然后循环这个集合调用future的get()取出数据。幸运的是completionservice帮你做了这件事情。
completionservice整合了executor和blockingqueue的功能。你可以将callable任务提交给它去执行，然后使用类似于队列中的take和poll方法，在结果完整可用时获得这个结果，像一个打包的future。
completionservice的take返回的future是哪个先完成就先返回哪一个，而不是根据提交顺序。

2 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任务。

然后看一下 completionservice.submit 方法：

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

可以看到，callable任务被包装成queueingfuture，而 queueingfuture是 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);
    }
}

可以看到在该 futuretask 中执行run方法，最终回调自定义的callable中的call方法，执行结束之后，

通过 set(result) 处理执行结果：

    /**
     * sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * <p>this method is invoked internally by the {@link #run} method
     * upon successful completion of the computation.
     *
     * @param v the value
     */
    protected void set(v v) {
        if (unsafe.compareandswapint(this, stateoffset, new, completing)) {
            outcome = v;
            unsafe.putorderedint(this, stateoffset, normal); // final state
            finishcompletion();
        }
    }

继续跟进finishcompletion()方法，在该方法中找到 done()方法：

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

可以看到该方法只做了一件事情，就是将执行结束的task添加到了队列中，只要队列中有元素，我们调用take()方法时就可以获得执行的结果。
到这里就已经清晰了，异步非阻塞获取执行结果的实现原理其实就是通过队列来实现的，futuretask将执行结果放到队列中，先进先出，线程执行结束的顺序就是获取结果的顺序。

completionservice实际上可以看做是executor和blockingqueue的结合体。completionservice在接收到要执行的任务时，通过类似blockingqueue的put和take获得任务执行的结果。completionservice的一个实现是executorcompletionservice，executorcompletionservice把具体的计算任务交给executor完成。

在实现上，executorcompletionservice在构造函数中会创建一个blockingqueue（使用的基于链表的无界队列linkedblockingqueue），该blockingqueue的作用是保存executor执行的结果。当计算完成时，调用futuretask的done方法。当提交一个任务到executorcompletionservice时，首先将任务包装成queueingfuture，它是futuretask的一个子类，然后改写futuretask的done方法，之后把executor执行的计算结果放入blockingqueue中。

queueingfuture的源码如下：

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

3 completionservice实现任务

public class completionservicetest {
    public static void main(string[] args) {

        executorservice threadpool = executors.newfixedthreadpool(10);
        completionservice<integer> completionservice = new executorcompletionservice<integer>(threadpool);
        for (int i = 1; i <=10; i++) {
            final int seq = i;
            completionservice.submit(new callable<integer>() {
                @override
                public integer call() throws exception {

                    thread.sleep(new random().nextint(5000));

                    return seq;
                }
            });
        }
        threadpool.shutdown();
        for (int i = 0; i < 10; i++) {
            try {
                system.out.println(
                        completionservice.take().get());
            } catch (interruptedexception e) {
                e.printstacktrace();
            } catch (executionexception e) {
                e.printstacktrace();
            }
        }

    }
}

7
3
9
8
1
2
4
6
5
10

4 completionservice总结

相比executorservice，completionservice可以更精确和简便地完成异步任务的执行
completionservice的一个实现是executorcompletionservice，它是executor和blockingqueue功能的融合体，executor完成计算任务，blockingqueue负责保存异步任务的执行结果
在执行大量相互独立和同构的任务时，可以使用completionservice
completionservice可以为任务的执行设置时限，主要是通过blockingqueue的poll(long time,timeunit unit)为任务执行结果的取得限制时间，如果没有完成就取消任务