Loading [MathJax]/jax/output/CommonHTML/config.js
首页
学习
活动
专区
圈层
工具
发布
首页
学习
活动
专区
圈层
工具
MCP广场
社区首页 >专栏 >多线程获取结果还在使用Future轮询获取结果吗?CompletionService快来了解下吧。

多线程获取结果还在使用Future轮询获取结果吗?CompletionService快来了解下吧。

原创
作者头像
java金融
修改于 2020-07-09 02:00:20
修改于 2020-07-09 02:00:20
1.5K00
代码可运行
举报
文章被收录于专栏:java金融java金融
运行总次数:0
代码可运行

背景

二胖上次写完参数校验(《二胖写参数校验的坎坷之路》)之后,领导一直不给他安排其他开发任务,就一直让他看看代码熟悉业务。二胖每天上班除了偶尔跟坐在隔壁的前端小姐姐聊聊天,就是看看这写枯燥无味的业务代码,无聊的一匹。虽然二胖已是久经职场的老油条了,但是看到同事们的周报都写的满满的,而自己的周报,就一两行,熟悉了什么功能。心里还是慌得一匹,毕竟公司不养闲人啊。于是乎二胖终于鼓起勇气为了向领导表明自己的上进心,主动向领导要开发任务。领导一看这小伙子这么有上进心,于是就到任务看板里面挑了一个业务逻辑比较简单的任务分配给了二胖。二胖拿到这个任务屁颠屁颠的回到座位。任务比较简单,就是通过爬虫去爬取某些卖机票(某猪、某携、某团等)的网站的一些机票,然后保存到数据库

同步入库

二胖拿到任务,三下五除二就把任务完成了。

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
 public static void main(String[] args) throws InterruptedException {
        String mouZhuFlightPrice = getMouZhuFlightPrice();
        String mouXieFlightPrice = getMouXieFlightPrice();
        String mouTuanFlightPrice = getMouTuanFlightPrice();
        saveDb(mouZhuFlightPrice);
        saveDb(mouXieFlightPrice);
        saveDb(mouTuanFlightPrice);
    }


    /**
     * 模拟请求某猪网站 爬取机票信息
     *
     *
     * @return
     * @throws InterruptedException
     */
    public static String getMouZhuFlightPrice() throws InterruptedException {
        // 模拟请求某猪网站 爬取机票信息
        Thread.sleep(10000);
        return "获取到某猪网站的机票信息了";
    }

    /**
     * 模拟请求某携网站 爬取机票信息
     *
     * @return
     * @throws InterruptedException
     */
    public static String getMouXieFlightPrice() throws InterruptedException {
        // 模拟请求某携网站 爬取机票信息
        Thread.sleep(5000);
        return "获取到某携网站的机票信息了";
    }


    /**
     * 模拟请求团网站 爬取机票信息
     *
     * @return
     * @throws InterruptedException
     */
    public static String getMouTuanFlightPrice() throws InterruptedException {
        // 模拟请求某团网站 爬取机票信息
        Thread.sleep(3000);
        return "获取到某团网站的机票信息了";
    }

    /**
     * 保存DB
     *
     * @param flightPriceList
     */
    public static void saveDb(String flightPriceList) {
            // 解析字符串 进行异步入库
    }

这次二胖学乖了,任务完成了先去找下坐他对面的技术大拿(看他那发际线就知道了)同事“二狗”让二狗大拿帮忙指点一二,看看代码是否还能有优化的地方。毕竟领导对代码的性能、以及代码的优雅是有要求的。领导多次在部门的周会上提到让我们多看看“二狗”写的代码,学习下人家写代码的优雅、抽象、封装等等。二狗大概的瞄了下二胖写的代码,提出了个小小的建议“这个代码可以采用多线程来优化下哦,你看某猪(CVM)这个网站耗时是拿到结果需要10s,其他的耗时都比它短,先有结果的我们可以先处理的,不需要等到大家都返回了再来处理的”。

轮循futureList获取结果

幸好二胖对多线程了解一点点,于是乎采用future的方式来实现。二胖使用一个List来保存每个任务返回的Future,然后去轮询这些Future,直到每个Future都已完成。由于需要先完成的任务需要先执行,且不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况,所以在调用get方式时,需要将超时时间设置为0

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
  public static void main(String[] args) {
        int taskSize = 3;
        Future<String> mouZhuFlightPriceFuture = executor.submit(() -> getMouZhuFlightPrice());
        Future<String> mouXieFlightPriceFuture = executor.submit(() -> getMouXieFlightPrice());
        Future<String> mouTuanFlightPriceFuture = executor.submit(() -> getMouTuanFlightPrice());
        List<Future<String>> futureList = new ArrayList<>();
        futureList.add(mouZhuFlightPriceFuture);
        futureList.add(mouXieFlightPriceFuture);
        futureList.add(mouTuanFlightPriceFuture);
        // 轮询,获取完成任务的返回结果
        while (taskSize > 0) {
            for (Future<String> future : futureList) {
                String result = null;
                try {
                    result = future.get(0, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    taskSize--;
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    taskSize--;
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    // 超时异常需要忽略,因为我们设置了等待时间为0,只要任务没有完成,就会报该异常
                }
                // 任务已经完成
                if (result != null) {
                    System.out.println("result=" + result);
                    // 从future列表中删除已经完成的任务
                    futureList.remove(future);
                    taskSize--;
                    // 此处必须break,否则会抛出并发修改异常。(也可以通过将futureList声明为CopyOnWriteArrayList类型解决)
                    break; // 进行下一次while循环
                }
            }
        }
    }

上述代码有两个小细节需要注意下:

  • 如采用ArrayList的话futureList删除之后需要break进行下一次while循环,否则会产生我们意想不到的ConcurrentModificationException异常。具体原因可看下《ArrayList的删除姿势你都掌握了吗》这个文章,里面有详细的介绍。
  • 在捕获了InterruptedExceptionExecutionException异常后记得 taskSize--否则就会发生死循环。如果生产发生了死循环你懂的,cpu被你打满,程序假死等。你离被开除也不远了。
  • 上面轮询future列表非常的复杂,而且还有很多异常需要处理,还有很多细节需要考虑,还有被开除的风险。所以这种方案也被pass了。

自定义BlockingQueue实现

  • 上述方案被pass之后,二胖就在思考可以借用哪种数据来实现下先进先出的功能,貌似队列可以实现下这个功能。所以二胖又写了一版采用队列来实现的功能。
代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
  final static ExecutorService executor = new ThreadPoolExecutor(6, 6,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Future<String> mouZhuFlightPriceFuture = executor.submit(() -> getMouZhuFlightPrice());
        Future<String> mouXieFlightPriceFuture = executor.submit(() -> getMouXieFlightPrice());
        Future<String> mouTuanFlightPriceFuture = executor.submit(() -> getMouTuanFlightPrice());

        // 创建阻塞队列
        BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(3);
        executor.execute(() -> run(mouZhuFlightPriceFuture, blockingQueue));
        executor.execute(() -> run(mouXieFlightPriceFuture, blockingQueue));
        executor.execute(() -> run(mouTuanFlightPriceFuture, blockingQueue));
        // 异步保存所有机票价格
        for (int i = 0; i < 3; i++) {
            String result = blockingQueue.take();
            System.out.println(result);
            saveDb(result);
        }
    }

    private static void run(Future<String> flightPriceFuture, BlockingQueue<String> blockingQueue) {
        try {
            blockingQueue.put(flightPriceFuture.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
  • 这次比上个版本好多了,代码也简洁多了。不过按理说这种需求应该是大家经常遇到的,应该不需要自己来实现把,JAVA这么贴心的语言应该会有api可以直接拿来用吧。

CompletionService实现

  • 二胖现在毕竟也是对代码的简洁性有追求的人了。于是乎二胖去翻翻自己躺在书柜里吃灰的并发相关的书籍,看看是否有解决方案。

终于皇天不负有心人在二胖快要放弃的时候突然发现了新大陆。 Java并发编程实战》一书6.3.5CompletionService:ExecutorBlockingQueue,有这样一段话:

如果向Executor提交了一组计算任务,并且希望在计算完成后获得结果,那么可以保留与每个任务关联的Future,然后反复使用get方法,同时将参数timeout指定为0,从而通过轮询来判断任务是否完成。这种方法虽然可行,但却有些繁琐。幸运的是,还有一种更好的方法:完成服务CompletionService。

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
  final static ExecutorService executor = new ThreadPoolExecutor(6, 6,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletionService completionService = new ExecutorCompletionService(executor);
        completionService.submit(() -> getMouZhuFlightPrice());
        completionService.submit(() -> getMouXieFlightPrice());
        completionService.submit(() -> getMouTuanFlightPrice());
        for (int i = 0; i < 3; i++) {
            String result = (String)completionService.take().get();
            System.out.println(result);
            saveDb(result);
        }
    }

当我们使用了CompletionService不用遍历future列表,也不需要去自定义队列了,代码变得简洁了。下面我们就来分析下CompletionService实现的原理吧。

CompletionService 介绍
  • 我们可以先看下JDK源码中CompletionServicejavadoc说明吧
代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
/**
 * 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.

大概意思是CompletionService实现了生产者提交任务和消费者获取结果的解耦,生产者和消费者都不用关心任务的完成顺序,由CompletionService来保证,消费者一定是按照任务完成的先后顺序来获取执行结果。

成员变量

既然需要按照任务的完成顺序获取结果,那内部应该也是通过队列来实现的吧。打开源码我们可以看到,里面有三个成员变量

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
public class ExecutorCompletionService<V> implements CompletionService<V> {
 // 执行task的线程池,创建CompletionService必须指定;
    private final Executor executor;
    //主要用于创建待执行task;
    private final AbstractExecutorService aes;
    //存储已完成状态的task,默认是基于链表结构的阻塞队列LinkedBlockingQueue。     
    private final BlockingQueue<Future<V>> completionQueue;
任务提交

ExecutorCompletionService任务的提交和执行都是委托给Executor来完成。当提交某个任务时,该任务首先将被包装为一个QueueingFuture

代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
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;
    }
任务完成后何时进入队列

从源码可以看出,QueueingFutureFutureTask的子类,实现了done方法,在task执行完成之后将当前task添加到completionQueue,将返回结果加入到阻塞队列中,加入的顺序就是任务完成的先后顺序。done方法的具体调用在FutureTaskfinishCompletion方法。

获取已完成任务
代码语言:javascript
代码运行次数:0
运行
AI代码解释
复制
 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);
    }

takepoll都是调用BlockingQueue提供的方法。

  • take() 获取任务阻塞,直到可以拿到任务为止。
  • poll() 获取任务不阻塞,如果没有获取到任务直接返回null
  • poll(long timeout, TimeUnit unit) 带超时时间等待的获取任务方法(一般推荐使用这种

总结

  • CompletionService 把线程池 Executor 和阻塞队列 BlockingQueue融合在一起,能够让批异步任务的管理更简单,将生产者提交任务和消费者获取结果的解耦。
  • CompletionService 能够让异步任务的执行结果有序化,先执行完的先进入阻塞队列,利用这个特性,我们可以轻松实现后续处理的有序性,避免无谓的等待。

结束

  • 由于自己才疏学浅,难免会有纰漏,假如你发现了错误的地方,还望留言给我指出来,我会对其加以修正。
  • 如果你觉得文章还不错,你的转发、分享、赞赏、点赞、留言就是对我最大的鼓励。
  • 感谢您的阅读,十分欢迎并感谢您的关注。

参考 《java并发编程实战》 https://www.jianshu.com/p/19093422dd57 https://blog.csdn.net/cbjcry/article/details/84222853 https://www.jianshu.com/p/493ae1b107e4

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
暂无评论
推荐阅读
编辑精选文章
换一批
聊聊 Java 8 CompletionService
CompletionService的实现目标是任务先完成可优先获取到,即结果按照完成先后顺序排序。
码猿技术专栏
2023/05/01
3390
聊聊 Java 8 CompletionService
ExecutorCompletionService 源码分析
概要 在ExecutorService的submit方法中可以获取返回值,通过Future的get方法,但是这个Future类存在缺陷,Future接口调用get()方法取得处理后的返回结果时具有阻塞性,也就是说调用Future的get方法时,任务没有执行完成,则get方法要一直阻塞等到任务完成为止。 这样大大的影响了系统的性能,这就是Future的最大缺点。为此,java1.5以后提供了CompletionServlice来解决这个问题。 CompletionService 接口CompletionSer
java404
2018/05/18
6660
并发编程 | CompletionService - 如何优雅地处理批量异步任务
上一篇文章中,我们详细地介绍了 CompletableFuture,它是一种强大的并发工具,能帮助我们以声明式的方式处理异步任务。虽然 CompletableFuture 很强大,但它并不总是最适合所有场景的解决方案。
kfaino
2023/11/02
1.5K0
【Java多线程-4】CompletionService详解
我们知道,通过 Future 和 FutureTask 可以获得线程任务的执行结果,但它们有一定的缺陷:
云深i不知处
2020/09/16
8030
(79) 方便的CompletionService / 计算机程序的思维逻辑
上节,我们提到,在异步任务程序中,一种常见的场景是,主线程提交多个异步任务,然后希望有任务完成就处理结果,并且按任务完成顺序逐个处理,对于这种场景,Java并发包提供了一个方便的方法,使用CompletionService,这是一个接口,它的实现类是ExecutorCompletionService,本节我们就来探讨它们。 基本用法 接口和类定义 与77节介绍的ExecutorService一样,CompletionService也可以提交异步任务,它的不同是,它可以按任务完成顺序获取结果,其具体定义为:
swiftma
2018/01/31
6240
【小家Java】Future、FutureTask、CompletionService、CompletableFuture解决多线程并发中归集问题的效率对比
开启线程执行任务,不管是使用Runnable(无返回值不支持上报异常)还是Callable(有返回值支持上报异常)接口,都可以轻松实现。那么如果是开启线程池并需要获取结果归集的情况下,如何实现,以及优劣?
YourBatman
2019/09/03
2.8K1
【小家Java】Future、FutureTask、CompletionService、CompletableFuture解决多线程并发中归集问题的效率对比
CompletionService学习
前面已经说到Future的默认实现是FutureTask,因此你可以看到其在jdk1.5的时候采用的是AQS去实现的,因此具有阻塞性,但jdk1.6之后,可以看到其基于CAS实现的。之所以学习Future,除了其具备异步功能,同时其采用的思想也是在设计模式中有体现的,也即Future模式,而且可以在kafka源码中看到基于Future构建的异步编程。
路行的亚洲
2020/07/16
2740
同事多线程使用不当导致OOM,被我怼了一顿
点击上方“芋道源码”,选择“设为星标” 管她前浪,还是后浪? 能浪的浪,才是好浪! 每天 10:33 更新文章,每天掉亿点点头发... 源码精品专栏 原创 | Java 2021 超神之路,很肝~ 中文详细注释的开源项目 RPC 框架 Dubbo 源码解析 网络应用框架 Netty 源码解析 消息中间件 RocketMQ 源码解析 数据库中间件 Sharding-JDBC 和 MyCAT 源码解析 作业调度中间件 Elastic-Job 源码解析 分布式事务中间件 TCC-Transaction
芋道源码
2022/05/09
6680
同事多线程使用不当导致OOM,被我怼了一顿
看完,你也能用多线程让接口提升5倍速!
酒店提供给各个渠道商房间价格是不一样的,我们需要轮询所有的渠道商接口,给用户返回一个最低的价格,前端会将这个价格显示给用户。
林老师带你学编程
2021/12/07
5830
Future和Callable学习
我们知道使用多线程时,最初的Thread到线程池,此时对于线程的使用,提供了其使用的复用率。而实现多线程的三种方式:继承Thread;实现Runnable接口,重写run方法;实现Callable接口,同时重写call方法,同时通过Future获取执行的返回值。也就是说callable执行任务,而Future拿到执行的结果。Future具有阻塞性在于其get()方法具有阻塞性,而isDone()是不具有阻塞性的。
路行的亚洲
2020/07/16
5130
高并发编程-ExecutorCompletionService深入解析
假设现在有一大批需要进行计算的任务,为了提高整批任务的执行效率,你可能会使用线程池,向线程池中不断submit异步计算任务,同时你需要保留与每个任务关联的Future,最后遍历这些Future,通过调用Future接口实现类的get方法获取整批计算任务的各个结果。
JavaQ
2019/06/11
9.6K0
高并发编程-ExecutorCompletionService深入解析
ExecutorCompletionService详解
ExecutorCompletionService 是Java并发编程中的一个有用的工具类,它实现了 CompletionService 接口。ExecutorCompletionService 将 Executor 和BlockingQueue 功能融合在一起,使用它可以提交我们的任务。这个任务委托给 Executor 执行,可以使用 ExecutorCompletionService 对象的 take() 和 poll() 方法获取结果。
BookSea
2024/01/04
4870
ExecutorCompletionService详解
CompletionService实践
在之前的ExecutorCompletionService中,我们了解了其实现的原理。通过FutureTask对提交任务的封装和代理然后通过call方法进行回调并将返回值存储在FutureTask中的Object中,当获取值的时候采用阻塞队列方式进行存储。其中的阻塞队列是对FutureTask的引用,也就是说提交的任务是有序的。获取到的值也是有顺序的。
写一点笔记
2020/09/01
5840
CompletionService实践
java中CompletionService的使用
之前的文章中我们讲到了ExecutorService,通过ExecutorService我们可以提交一个个的task,并且返回Future,然后通过调用Future.get方法来返回任务的执行结果。
程序那些事
2020/07/08
9200
【并发那些事】线程有序化神器CompletionService
话说有一天,产品经理突然找到正在摸鱼的你。 产品:『我们要加一个聚合搜索功能,当用户在我们网站查询一件商品时,我们分别从 A、B、C 三个网站上查询这个信息,然后再把得到的结果返回给用户』 你:『哦,就是写个爬虫,从 3 个网站上抓取数据是吧?』 产品:『呸,爬虫是犯法的,这叫数据分析,怎么样,能实现吧?』 你:『可以』 产品:『好的,明天上线』 你:『。。。』
Bug开发工程师
2020/03/02
5090
【并发那些事】线程有序化神器CompletionService
“既生 ExecutorService, 何生 CompletionService?”
前言 在 我会手动创建线程,为什么要使用线程池? 中详细的介绍了 ExecutorService,可以将整块任务拆分做简单的并行处理; 在 不会用Java Future,我怀疑你泡茶没我快 中又详细的
用户4172423
2020/08/13
7660
“既生 ExecutorService, 何生 CompletionService?”
(juc系列)completionservice源码阅读
看到这个,其实已经基本明了了. 任务执行完成后,将任务放入到内部的阻塞队列中,那么获取时就是按照任务完成顺序了。
呼延十
2021/10/22
2880
CompletionService小技巧
在上一篇blogs中,我详细的解释了CompletionService的使用方法和ExecutorCompletionService的详细实现,这篇blogs中,我就介绍使用它的一个小技巧,算是对上一篇blogs的一个补完。在开始之前我们先回顾一下它的实现。
Allen Cheng
2018/09/10
9220
教校花学妹JDK批量异步任务最强工具CompletionService
如何优化一个询价应用的核心代码?如果采用“ThreadPoolExecutor+Future”,优化代码可能如下:
JavaEdge
2021/04/19
8030
教校花学妹JDK批量异步任务最强工具CompletionService
java高并发系列 - 第20天:JUC中的Executor框架详解2
买新房了,然后在网上下单买冰箱、洗衣机,电器商家不同,所以送货耗时不一样,然后等他们送货,快递只愿送到楼下,然后我们自己将其搬到楼上的家中。
路人甲Java
2019/12/10
5190
相关推荐
聊聊 Java 8 CompletionService
更多 >
LV.1
公众号【java金融】java资深软件工程师
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档
本文部分代码块支持一键运行,欢迎体验
本文部分代码块支持一键运行,欢迎体验