java利用CompletionService保证任务先完成先获取到执行结果

ExecutorService executorService = Executors.newFixedThreadPool(4);
List<Future> futures = new ArrayList<Future<Integer>>();
futures.add(executorService.submit(A));
futures.add(executorService.submit(B));
futures.add(executorService.submit(C));
// 遍历 Future list，通过 get() 方法获取每个 future 结果
for (Future future:futures) {
	Integer result = future.get();
	// 其他业务逻辑 如果A执行时间很长，阻塞
}

ExecutorService executorService = Executors.newFixedThreadPool(4);
// ExecutorCompletionService 是 CompletionService 唯一实现类
CompletionService completionService = new ExecutorCompletionService<>(executorService );
List<Future> futures = new ArrayList<Future<Integer>>();
futures.add(completionService.submit(A));
futures.add(completionService.submit(B));
futures.add(completionService.submit(C));
// 遍历 Future list，通过 get() 方法获取每个 future 结果
for (int i = 0; i < futures.size(); i++) {
    Integer result = completionService.take().get();
    // 其他业务逻辑
}

// 创建阻塞队列
BlockingQueue<Integer> bq =
  new LinkedBlockingQueue<>();
// 任务A 异步进入阻塞队列  
executor.execute(() -> bq.put(A.get()));
// 任务B 异步进入阻塞队列  
executor.execute(() -> bq.put(B.get()));
// 任务C 异步进入阻塞队列  
executor.execute(()-> bq.put(C.get()));
for (int i = 0; i < 3; i++) {
  Integer r = bq.take();
  // 异步执行所有业务逻辑
  executor.execute(()->action(r));
}

public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take() throws InterruptedException;
    Future<V> poll();
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}

public ExecutorCompletionService(Executor executor) {
    if (executor == null)
        throw new NullPointerException();
    this.executor = executor;
    // 判断executor是不是ThreadPoolExecutor,ScheduledThreadPoolExecutor,ForkJoinPool
    // 其余框架也有实现了AbstractExecutorService抽象类，目前JDK里只有上述的三种实现
    // 如果不是，则为null
    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;
    // 判断executor是不是ThreadPoolExecutor,ScheduledThreadPoolExecutor,ForkJoinPool
    // 其余框架也有实现了AbstractExecutorService抽象类，目前JDK里只有上述的三种实现
    // 如果不是，则为null
    this.aes = (executor instanceof AbstractExecutorService) ?
        (AbstractExecutorService) executor : null;
    this.completionQueue = completionQueue;
}

public Future<V> submit(Callable<V> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<V> f = newTaskFor(task);
    executor.execute(new QueueingFuture<V>(f, completionQueue));
    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<V>(f, completionQueue));
    return f;
}

private RunnableFuture<V> newTaskFor(Callable<V> task) {
    // aes是AbstractExecutorService，其实现类是ThreadPoolExecutor，ForkJoinPool，SchedulerThreadPoolExecutor
    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);
}

private static class QueueingFuture<V> extends FutureTask<Void> {
    QueueingFuture(RunnableFuture<V> task,
                   BlockingQueue<Future<V>> completionQueue) {
        super(task, null);
        this.task = task;
        this.completionQueue = completionQueue;
    }
    private final Future<V> task;
    private final BlockingQueue<Future<V>> completionQueue;
    // 会被java.util.concurrent.FutureTask#finishCompletion调用，判读是否计算完成
    // 计算结果放在阻塞队列中
    protected void done() { completionQueue.add(task); }
}

// 从结果队列中获取并移除一个已经执行完成的任务的结果，如果没有就会阻塞，直到有任务完成返回结果。
public Future<V> take() throws InterruptedException {
    return completionQueue.take();
}
// 从结果队列中获取并移除一个已经执行完成的任务的结果，如果没有就会返回null，该方法不会阻塞。
public Future<V> poll() {
    return completionQueue.poll();
}
// 从结果队列中获取并移除一个已经执行完成的任务的结果，如果没有就会返回null，该方法不会阻塞。
// 超时
public Future<V> poll(long timeout, TimeUnit unit)
        throws InterruptedException {
    return completionQueue.poll(timeout, unit);
}