CompletionService的主要作用是：按照异步任务的完成顺序，逐个获取到已经完成的异步任务。主要实现是在ExecutorCompletionService中。

类图

ExecutorCompletionService.png

核心内部类

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

在CompletionService的实现中，将任务FutureTask做了扩展，实现了FutureTask的done方法。当任务完成后会回调这个方法，这时我们在这个方法中将完成的任务放到队列中，就实现了按照异步任务完成的顺序，逐个处理任务的结果了。

核心属性

// 执行任务的线程池
private final Executor executor;
// 存放已完成的异步任务的阻塞队列，默认使用 LinkedBlockingQueue
private final BlockingQueue<Future<V>> completionQueue;

构造函数

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线程池的实现来执行异步任务，但是建议再传入一个阻塞队列，默认的LinkedBlockingQueue是一个无界队列，有内存溢出的风险。

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

我们可以看到，在提交任务给线程池之前，我们会将任务封装成QueueingFuture任务。当该任务执行完成后会回调执行done方法，将任务放到队列。

获取已完成的任务

public Future<V> take() throws InterruptedException {
    return completionQueue.take();
}

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

• take：如果没有任务，一直阻塞，直到有新任务进来
• poll：如果没有任务返回NULL

示例

public class CompletionServiceTest {

    @Test
    public void test() throws ExecutionException, InterruptedException {
        Random random = new Random();
        ExecutorService executor = Executors.newFixedThreadPool(10);
        CompletionService<Integer> completionService = new ExecutorCompletionService<>(executor, new LinkedBlockingQueue<>(10));
        for (int i = 0; i < 8; i++) {
            completionService.submit(() -> {
                int time = random.nextInt(1000);
                sleep(time);
                System.out.println(Thread.currentThread().getName() + " 执行异步任务执行耗时： " + time);
                return time;
            });
        }

        while (true) {
            System.out.println(Thread.currentThread().getName() + " 主线程获取到任务结果 " + completionService.take().get());
        }
    }


    public static void sleep(int probe) {
        try {
            Thread.sleep(probe);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

pool-1-thread-7 执行异步任务执行耗时： 153
main 主线程获取到任务结果 153
pool-1-thread-5 执行异步任务执行耗时： 208
main 主线程获取到任务结果 208
pool-1-thread-4 执行异步任务执行耗时： 242
main 主线程获取到任务结果 242
pool-1-thread-8 执行异步任务执行耗时： 456
main 主线程获取到任务结果 456
pool-1-thread-1 执行异步任务执行耗时： 567
main 主线程获取到任务结果 567
pool-1-thread-2 执行异步任务执行耗时： 782
main 主线程获取到任务结果 782
pool-1-thread-6 执行异步任务执行耗时： 796
main 主线程获取到任务结果 796
pool-1-thread-3 执行异步任务执行耗时： 976
main 主线程获取到任务结果 976

我的是8核机器，所以任务的结束时间一定会按照任务的结束时间排序。

源码

https://github.com/wyh-spring-ecosystem-student/spring-boot-student/tree/releases

spring-boot-student-concurrent 工程

layering-cache

为监控而生的多级缓存框架 layering-cache这是我开源的一个多级缓存框架的实现，如果有兴趣可以看一下