(79) 方便的CompletionService / 计算机程序的思维逻辑

上节，我们提到，在异步任务程序中，一种常见的场景是，主线程提交多个异步任务，然后希望有任务完成就处理结果，并且按任务完成顺序逐个处理，对于这种场景，Java并发包提供了一个方便的方法，使用CompletionService，这是一个接口，它的实现类是ExecutorCompletionService，本节我们就来探讨它们。 基本用法 接口和类定义 与77节介绍的ExecutorService一样，CompletionService也可以提交异步任务，它的不同是，它可以按任务完成顺序获取结果，其具体定义为：

public interface CompletionService<V> { Future<V> submit(Callable<V> task); Future<V> submit(Runnable task, V result); Future<V> take() throws InterruptedException; Future<V> poll(); Future<V> 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<Future<V>> completionQueue)

至少需要一个Executor参数，可以提供一个BlockingQueue参数，用作完成任务的队列，没有提供的话，ExecutorCompletionService内部会创建一个LinkedBlockingQueue。 基本示例 我们在77节的invokeAll的示例中，演示了并发下载并分析URL的标题，那个例子中，是要等到所有任务都完成才处理结果的，这里，我们修改一下，一有任务完成就输出其结果，代码如下：

public class CompletionServiceDemo { static class UrlTitleParser implements Callable<String> { 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<String> urls) throws InterruptedException { ExecutorService executor = Executors.newFixedThreadPool(10); try { CompletionService<String> completionService = new ExecutorCompletionService<>( executor); for (String url : urls) { completionService.submit(new UrlTitleParser(url)); } for (int i = 0; i < urls.size(); i++) { Future<String> 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<String> 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方法，该方法代码为：

protected void done() { }

它的实现为空，但它是一个protected方法，子类可以重写该方法。 在ExecutorCompletionService中，提交的任务类型不是一般的FutureTask，而是一个子类QueueingFuture，如下所示：

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

该子类重写了done方法，在任务完成时将结果加入到完成队列中，其代码为：

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

ExecutorCompletionService的take/poll方法就是从该队列获取结果，如下所示：

public Future<V> take() throws InterruptedException { return completionQueue.take(); }

实现invokeAny 我们在77节提到，AbstractExecutorService的invokeAny的实现，就利用了ExecutorCompletionService，它的基本思路是，提交任务后，通过take方法获取结果，获取到第一个有效结果后，取消所有其他任务，不过，它的具体实现有一些优化，比较复杂。我们看一个模拟的示例，从多个搜索引擎查询一个关键词，但只要任意一个的结果就可以，模拟代码如下：

public class InvokeAnyDemo { static class SearchTask implements Callable<String> { private String engine; private String keyword; public SearchTask(String engine, String keyword) { this.engine = engine; this.keyword = keyword; } @Override public String call() throws Exception { // 模拟从给定引擎搜索结果 Thread.sleep(engine.hashCode() % 1000); return "<result for> " + keyword; } } public static String search(List<String> engines, String keyword) throws InterruptedException { ExecutorService executor = Executors.newFixedThreadPool(10); CompletionService<String> cs = new ExecutorCompletionService<>(executor); List<Future<String>> futures = new ArrayList<Future<String>>( engines.size()); String result = null; try { for (String engine : engines) { futures.add(cs.submit(new SearchTask(engine, keyword))); } for (int i = 0; i < engines.size(); i++) { try { result = cs.take().get(); if (result != null) { break; } } catch (ExecutionException ignore) { // 出现异常，结果无效，继续 } } } finally { // 取消所有任务，对于已完成的任务，取消没有什么效果 for (Future<String> f : futures) f.cancel(true); executor.shutdown(); } return result; } public static void main(String[] args) throws InterruptedException { List<String> engines = Arrays.asList(new String[] { "www.baidu.com", "www.sogou.com", "www.so.com", "www.google.com" }); System.out.println(search(engines, "老马说编程")); } }

SearchTask模拟从指定搜索引擎查询结果，search利用CompletionService/ExecutorService执行并发查询，在得到第一个有效结果后，取消其他任务。 小结 本节比较简单，主要就是介绍了CompletionService的用法和原理，它通过一个额外的结果队列，方便了对于多个异步任务结果的处理。 下一节，我们来探讨一种常见的需求 - 定时任务。

(与其他章节一样，本节所有代码位于 https://github.com/swiftma/program-logic)