java - concurrent 之 CompletionService
博客专区 > yuzn 的博客 > 博客详情
java - concurrent 之 CompletionService
yuzn 发表于1年前
java - concurrent 之 CompletionService
  • 发表于 1年前
  • 阅读 8
  • 收藏 0
  • 点赞 0
  • 评论 0

腾讯云 成都新区开服特惠,热门产品8.5折>>>   

  • 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);
}
  •  
共有 人打赏支持
粉丝 14
博文 24
码字总数 13672
×
yuzn
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: