Java Thread&Concurrency(8): 深入理解CompletionService接口及其实现
2014-06-12 16:27
861 查看
背景(注释):
我们接着来看下CompletionService接口和它的实现ExecutorCompletionService。
CompletionService接口提供了一种在产生的异步任务和任务的计算结果之间解耦的服务。生产者通过submit提交任务。消费者通过take方法得到已经被执行的任务然后通过他们的完成次序来处理他们的结果。一个CompletionService能够在异步IO的场景中使用,这种情况下任务在程序的一个地方中提交,但是在另一个地方检查是否完成,完成的顺序与提交顺序不同。
一般情况下,CompletionService依赖于一个独立的Executor去实际地执行任务。CompletionService仅仅只是维护一个内部的完成队列。ExecutorCompletionService提供了一个这样的实现。
内存一致性:一个线程提交一个任务前的行为happen-before任务执行的行为happen-before通过take操作得到结果之后的行为。
ExecutorCompletionService类通过一个独立的Executor来执行任务。这个类将被提交的任务,在完成之后放入一个队列(通过tak方法取得)。这个类是足够轻量的(所以是短暂的),从而可以实现大量的任务处理。
使用场景:
假设你有一大群解决者来解决一个问题,每个返回一个结果Result,并且并发地执行任务。我们需要在其中的解决者完成任务后尽快处理结果(使用use),那么你可以使用如下示例:
假设我们需要处理第一个非null的任务结果,并且忽略所有异常和取消其他所有的任务(党第一个被准备好时):
实现原理:
这部分代码较简单,可以放下面一起说明:
executor用于实际上执行任务。
aes用于创建任务(一般情况下等同于executor对象)。
completionQueue为服务的完成队列。
这里的QueueingFuture作为FutureTask的一个扩展,它实际上只做一件事:在任务完成后把任务添加到服务私有的CompletionQueue完成队列中,然后就可以从中获取。
我们来看提交(submit)和获取(take)方法:
实际提交的是QueueingFuture对象。
提供了从中获取Future对象的接口(阻塞、非阻塞、限时阻塞)。
注意,这里得到的返回的是Future,并不是result,Future实际上就是submit中创建的RunnableFuture。
我们接着来看下CompletionService接口和它的实现ExecutorCompletionService。
CompletionService接口提供了一种在产生的异步任务和任务的计算结果之间解耦的服务。生产者通过submit提交任务。消费者通过take方法得到已经被执行的任务然后通过他们的完成次序来处理他们的结果。一个CompletionService能够在异步IO的场景中使用,这种情况下任务在程序的一个地方中提交,但是在另一个地方检查是否完成,完成的顺序与提交顺序不同。
一般情况下,CompletionService依赖于一个独立的Executor去实际地执行任务。CompletionService仅仅只是维护一个内部的完成队列。ExecutorCompletionService提供了一个这样的实现。
内存一致性:一个线程提交一个任务前的行为happen-before任务执行的行为happen-before通过take操作得到结果之后的行为。
ExecutorCompletionService类通过一个独立的Executor来执行任务。这个类将被提交的任务,在完成之后放入一个队列(通过tak方法取得)。这个类是足够轻量的(所以是短暂的),从而可以实现大量的任务处理。
使用场景:
假设你有一大群解决者来解决一个问题,每个返回一个结果Result,并且并发地执行任务。我们需要在其中的解决者完成任务后尽快处理结果(使用use),那么你可以使用如下示例:
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); } }
假设我们需要处理第一个非null的任务结果,并且忽略所有异常和取消其他所有的任务(党第一个被准备好时):
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); }
实现原理:
这部分代码较简单,可以放下面一起说明:
private final Executor executor; private final AbstractExecutorService aes; private final BlockingQueue<Future<V>> completionQueue; /** * FutureTask extension to enqueue upon completion */ 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; }三个基本的要素:
executor用于实际上执行任务。
aes用于创建任务(一般情况下等同于executor对象)。
completionQueue为服务的完成队列。
这里的QueueingFuture作为FutureTask的一个扩展,它实际上只做一件事:在任务完成后把任务添加到服务私有的CompletionQueue完成队列中,然后就可以从中获取。
我们来看提交(submit)和获取(take)方法:
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; } public Future<V> submit(Runnable task, V result) { if (task == null) throw new NullPointerException(); RunnableFuture<V> f = newTaskFor(task, result); executor.execute(new QueueingFuture(f)); return f; }
实际提交的是QueueingFuture对象。
public Future<V> take() throws InterruptedException { return completionQueue.take(); } public Future<V> poll() { return completionQueue.poll(); } public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException { return completionQueue.poll(timeout, unit); }
提供了从中获取Future对象的接口(阻塞、非阻塞、限时阻塞)。
注意,这里得到的返回的是Future,并不是result,Future实际上就是submit中创建的RunnableFuture。
相关文章推荐
- Java Thread&Concurrency(7): 深入理解Callable/Future(FutureTask)接口及其实现
- Java Thread&Concurrency(11): 深入理解ThreadPoolExecutor及其实现原理
- Java Thread&Concurrency(14): 深入理解条件队列(Condition)及其实现原理
- Java Thread&Concurrency(12): 深入理解AbstractExecutorService及其实现原理
- Java Thread&Concurrency(9): 深入理解StampedLock及其实现原理
- Java Thread&Concurrency(16): 深入理解ArrayBlockingQueue及其实现原理
- Java Thread&Concurrency(13): 深入理解ConcurrentLinkedQueue及其实现原理
- Java Thread&Concurrency(15): 深入理解ScheduledThreadPoolExecutor及其实现原理
- Java Thread&Concurrency(10): 深入理解ThreadLocal及其实现原理
- Java Thread&Concurrency(4): 深入理解Exchanger实现原理
- Java Thread&Concurrency(5): 深入理解Phaser实现原理
- Java Thread&Concurrency(2): 深入理解ConcurrentSkipListMap实现原理
- Java Thread&Concurrency(3): 深入理解SynchronousQueue实现原理
- Java Thread&Concurrency(1): 深入理解Fork-Join并发执行框架
- 对java中的接口和实现类的深入理解
- java之实现Cloneable接口的详解,克隆一个对象--对应有浅克隆和深克隆,概念结合代码深入理解
- 【Java多线程学习】Java多线程的两种实现方式:继承Thread类 & 实现Runable接口
- [java]深入理解java中的char,unicode编码及其实现方式
- 对java 接口和实现的理解
- java中的通过继承Thread和实现Runnable接口实现的线程