文档章节

java - concurrent 之 CompletionService

yuzn
 yuzn
发布于 2016/08/17 16:40
字数 704
阅读 9
收藏 0
  • 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);
}
  •  

© 著作权归作者所有

共有 人打赏支持
yuzn
粉丝 13
博文 31
码字总数 14730
作品 0
项目经理
JDK5多线程框架java.util.concurrent

一般的服务器都需要线程池,比如Web、FTP等服务器,不过它们一般都自己实现了线程池,比如以前介绍过的Tomcat、Resin和Jetty等,现在有了JDK5,我们就没有必要重复造车轮了,直接使用就可以,...

johnnyhg
2009/05/08
0
2
Java 并发 – 第七部分:Executors 与线程池

现在让我们开始 Java 并发系列的新篇章。这壹次我们会学习如何干净的启动壹個新线程,以及如何在线程池中管理它。在 Java 中,假设你有壹個像下面这样的 Runnable 线程: Runnable runnable ...

苗哥
2013/10/07
0
5
ThreadLocal in Java - Example Program and Tutorial

ThreadLocal in Java is another way to achieve thread-safety apart from writing immutable classes. If you have been writing multi-threaded or concurrent code in Java then you mus......

perfectspr
2014/12/10
0
0
Java系列笔记 - Java 内存区域和GC机制

Java垃圾回收概况   Java GC(Garbage Collection,垃圾收集,垃圾回收)机制,是Java与C++/C的主要区别之一,作为Java开发者,一般不需要专门编写内存回收和垃圾清理代码,对内存泄露和溢...

明舞
2015/08/18
0
6
Java Processes and Threads 进程与线程

Processes and Threads In concurrent programming, there are two basic units of execution: processes and threads. In the Java programming language, concurrent programming is mostl......

Oscarfff
2015/04/11
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

Zookeeper总结

Zookeeper的部分概念 什么是zookeeeper? Zookeeper是一个分布式服务的协调中心 zookeeper节点的角色类型? Leader(领导者)、Follower(跟随者)、Observer(观察者) Leader 负责更新系统...

DemonsI
24分钟前
1
0
Redis学习笔记

常用命令 从Docker进入Redis的命令 sudo docker exec -it redis /bin/bash

OSC_fly
25分钟前
0
0
SqlServer查询某个日期的数据

select * from View_ZJMONITORINGCORROSION where ENTERDATE > CONVERT(datetime,DATEADD(day,1,'2017/12/28 14:53:07'))...

笑丶笑
26分钟前
0
0
常用编码规范

Standard characters https://ascii.cl/

yeahlife
28分钟前
0
0
flannel实战

docker swarm mode的出现是个里程碑,官方原生的编排调度看起来都成雏形了,但是swarm mode和容器外部系统的对接、网络性能始终不尽人意,swarm mode下各种开源周边不能使用,感觉swarm mod...

China_OS
30分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部