查看历史文章,请点击上方链接关注公众号。
上节
,我们提到,在异步任务程序中,一种常见的场景是,主线程提交多个异步任务,然后希望有任务完成就处理结果,并且按任务完成顺序逐个处理,对于这种场景,Java并发包提供了一个方便的方法,使用CompletionService,这是一个接口,它的实现类是ExecutorCompletionService,本节我们就来探讨它们。
基本用法
接口和类定义
与
77节
介绍的ExecutorService一样,CompletionService也可以提交异步任务,它的不同是,它可以按任务完成顺序获取结果,其具体定义为:
public interface CompletionService
{
Future
submit(Callable
task);
Future
submit(Runnable task, V result);
Future
take() throws InterruptedException;
Future
poll();
Future
poll(long timeout, TimeUnit unit) throws InterruptedException;
}
其submit方法与ExecutorService是一样的,多了take和poll方法,它们都是获取下一个完成任务的结果,take()会阻塞等待,poll()会立即返回,如果没有已完成的任务,返回null,带时间参数的poll方法会最多等待限定的时间。
CompletionService的主要实现类是ExecutorCompletionService,它依赖于一个Executor完成实际的任务提交,而自己主要负责结果的排队和处理,它的构造方法有两个:
public ExecutorCompletionService(Executor executor)
public ExecutorCompletionService(Executor executor, BlockingQueue
> completionQueue)
至少需要一个Executor参数,可以提供一个BlockingQueue参数,用作完成任务的队列,没有提供的话,ExecutorCompletionService内部会创建一个LinkedBlockingQueue。
基本示例
我们在
77节
的invokeAll的示例中,演示了并发下载并分析URL的标题,那个例子中,是要等到所有任务都完成才处理结果的,这里,我们修改一下,一有任务完成就输出其结果,代码如下:
public class CompletionServiceDemo {
static class UrlTitleParser implements Callable
{
private String url;
public UrlTitleParser(String url) {
this.url = url;
}
@Override
public String call() throws Exception {
Document doc = Jsoup.connect(url).get();
Elements elements = doc.select("head title");
if (elements.size() > 0) {
return url + ": " + elements.get(0).text();
}
return null;
}
}
public static void parse(List
urls) throws InterruptedException {
ExecutorService executor = Executors.newFixedThreadPool(10);
try {
CompletionService
completionService = new ExecutorCompletionService<>(
executor);
for (String url : urls) {
completionService.submit(new UrlTitleParser(url));
}
for (int i = 0; i < urls.size(); i++) {
Future
result = completionService.take();
try {
System.out.println(result.get());
} catch (ExecutionException e) {
e.printStackTrace();
}
}
} finally {
executor.shutdown();
}
}
public static void main(String[] args) throws InterruptedException {
List
urls = Arrays.asList(new String[] {
"http://www.cnblogs.com/swiftma/p/5396551.html",
"http://www.cnblogs.com/swiftma/p/5399315.html",
"http://www.cnblogs.com/swiftma/p/5405417.html",
"http://www.cnblogs.com/swiftma/p/5409424.html" });
parse(urls);
}
}
在parse方法中,首先创建了一个ExecutorService,然后才是CompletionService,通过后者提交任务、按完成顺序逐个处理结果,这样,是不是很方便?
基本原理
ExecutorCompletionService是怎么让结果有序处理的呢?其实,也很简单,如前所述,它有一个额外的队列,每个任务完成之后,都会将代表结果的Future入队。
那问题是,任务完成后,怎么知道入队呢?我们具体来看下。
在
77节
我们介绍过FutureTask,任务完成后,不管是正常完成、异常结束、还是被取消,都会调用finishCompletion方法,而该方法会调用一个done方法,该方法代码为: