java - concurrent 之 CompletionService

• CompletionService

/**
 * A service that decouples the production of new asynchronous tasks
 * from the consumption of the results of completed tasks.  Producers
 * {@code submit} tasks for execution. Consumers {@code take}
 * completed tasks and process their results in the order they
 * complete.  A {@code CompletionService} can for example be used to
 * manage asynchronous I/O, in which tasks that perform reads are
 * submitted in one part of a program or system, and then acted upon
 * in a different part of the program when the reads complete,
 * possibly in a different order than they were requested.**/

参与java doc能够看到如上描述。简单来讲就是CompletionService使(批)任务异步执行与任务结果处理分离：即生产者执行任务，消费者处理任务结果；而且在后面描述一个在异步I/O上的一个使用场景。

• API

Future<V> submit(Callable<V> task);

@param result the result to return upon successful completion
Future<V> submit(Runnable task, V result);

/**
 * 此方法阻塞获取已完成的任务Future，并从任务列表中移除
 * @return the Future representing the next completed task
 * @throws InterruptedException if interrupted while waiting
 */
Future<V> take() throws InterruptedException;

/**
 * 比较take，此方法是非阻塞的，若是没有完成的任务，返回null
 */
Future<V> poll();

Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;

• 实现 ExecutorCompletionService

ExecutorCompletionService是CompletionService的惟一实现

private final Executor executor;
private final AbstractExecutorService aes;
private final BlockingQueue<Future<V>> completionQueue;

public ExecutorCompletionService(Executor executor) {

public ExecutorCompletionService(Executor executor, BlockingQueue<Future<V>> completionQueue)

每一个任务的提交都会构造一个QueueingFutrue

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

而QueueingFutrue有个回调方法，在任务执行完成后，放到completionQueue阻塞队列中

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

由此实现方式很明显了。

• Demo

查询文档能够看到官方提供的案例

/** Suppose you have a set of solvers for a certain problem, each
* returning a value of some type {@code Result}, and would like to
* run them concurrently, processing the results of each of them that
* return a non-null value, in some method {@code use(Result r)}. You
* could write this as:
* 大意就是：假设你有一批须要回执的任务要并发处理，就能够使用以下方式(ps: 也能够利用Futrue方式的实现
* ，这里再也不说明)
*/
 void solve(Executor e,
            Collection<Callable<Result>> solvers)
     throws InterruptedException, ExecutionException {
     CompletionService<Result> ecs
         = new ExecutorCompletionService<Result>(e);
     for (Callable<Result> s : solvers)
         ecs.submit(s);
     int n = solvers.size();
     for (int i = 0; i < n; ++i) {
         Result r = ecs.take().get();
         if (r != null)
             use(r);
     }
}

/** Suppose instead that you would like to use the first non-null result
* of the set of tasks, ignoring any that encounter exceptions,
* and cancelling all other tasks when the first one is ready:
* 大意是：若是只想获得率先执行完任务的返回值，忽略其余的任务执行状况，而且在第一个任务执行结束后取消其余任务
*/

 void solve(Executor e,
            Collection<Callable<Result>> solvers)
     throws InterruptedException {
     CompletionService<Result> ecs
         = new ExecutorCompletionService<Result>(e);
     int n = solvers.size();
     List<Future<Result>> futures
         = new ArrayList<Future<Result>>(n);
     Result result = null;
     try {
         for (Callable<Result> s : solvers)
             futures.add(ecs.submit(s));
         for (int i = 0; i < n; ++i) {
             try {
                 Result r = ecs.take().get();
                 if (r != null) {
                     result = r;
                     break;
                 }
             } catch (ExecutionException ignore) {}
         }
     }
    finally {
         for (Future<Result> f : futures)
             f.cancel(true);
     }

     if (result != null)
         use(result);
}