首页 > 代码库 > 计算机程序的思维逻辑 (79) - 方便的CompletionService

计算机程序的思维逻辑 (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)

----------------

未完待续,查看最新文章,敬请关注微信公众号“老马说编程”(扫描下方二维码),从入门到高级,深入浅出,老马和你一起探索Java编程及计算机技术的本质。用心原创,保留所有版权。

技术分享

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