java-concurrent
  • 前言
  • Java多线程基础
    • 线程简介
      • 什么是线程
      • 为什么要使用多线程/线程使用的好处
      • 线程的优先级
      • 线程的状态
      • Daemon线程
    • 启动和终止线程
      • 构造线程
      • 启动线程
      • 中断线程
      • 过期的suspend()、resume()和stop()
      • 安全地终止线程
    • 多线程实现方式
    • 多线程环境下,局部变量和全局变量都会共享吗?
    • Java线程间的协助和通信
      • Thread.join的使用
      • volatile、ThreadLocal、synchronized3个关键字区别
      • volatile关键字
      • ThreadLocal关键字
      • synchronized关键字
      • Java线程等待和通知的相关方法
    • 实战应用
      • 连接池
      • 线程池
      • 如何计算合适的线程数
  • Java线程池与框架
    • Executor 框架
    • 自定义线程池——ThreadPoolExecutor
    • 线程池工具类(单例模式)
    • 关闭线程池
    • 合理地配置线程池
    • 线程池的监控
    • RejectedExecutionException产生的原因
    • SpringBoot配置线程池工具类
    • FutureTask详解
    • CompletionService讲解
    • Future、FutureTask、CompletionService、CompletableFuture区别
  • Java内存模型
    • Java 内存模型的基础
      • 并发编程模型的两个关键问题
      • Java内存模型的抽象结构
      • 从源代码到指令序列的重排序
      • 并发编程模型的分类
    • 重排序
      • 数据依赖性
      • as-if-serial语义
      • 程序顺序规则
      • 重排序对多线程的影响
    • 顺序一致性
      • 数据竞争与顺序一致性
      • 顺序一致性内存模型
      • 同步程序的顺序一致性效果
      • 未同步程序的执行特性
    • volatile内存语义
      • volatile的特性
      • volatile写-读建立的happens-before关系
      • volatile写-读的内存语义
      • volatile内存语义的实现
      • JSR-133为什么要增强volatile的内存语义
    • 锁内存定义
      • 锁的释放-获取建立的happens-before关系
      • 锁的释放和获取的内存语义
      • 锁内存语义的实现
      • concurrent包的实现
    • final域内存语义
      • final域的重排序规则
      • 写final域的重排序规则
      • 读final域的重排序规则
      • final域为引用类型
      • 为什么final引用不能从构造函数内“溢出”
      • final语义在处理器中的实现
      • JSR-133为什么要增强final的语义
    • happens-before
    • 双重检查锁定与延迟初始化
      • 双重检查锁定的由来
      • 问题的根源
      • 基于volatile的解决方案
      • 基于类初始化的解决方案
    • Java内存模型综述
      • 处理器的内存模型
      • 各种内存模型之间的关系
      • JMM的内存可见性保证
      • JSR-133对旧内存模型的修补
  • HashMap实现原理
    • 讲解(一)
    • 讲解(二)
    • HashMap原理(面试篇)
    • HashMap原理(面试篇二)
  • ConcurrentHashMap的实现原理与使用
    • 为什么要使用ConcurrentHashMap
    • ConcurrentHashMap的结构
    • ConcurrentHashMap的初始化
    • 定位Segment
    • ConcurrentHashMap的操作
    • ConcurrentHashMap讲解(一)
  • Java中的阻塞队列
    • 什么是阻塞队列
    • Java里的阻塞队列
    • 阻塞队列的实现原理
  • Fork/Join框架
    • 什么是Fork/Join框架
    • 工作窃取算法
    • Fork/Join框架的设计
    • 使用Fork/Join框架
    • Fork/Join框架的异常处理
    • Fork/Join框架的实现原理
    • ForkJoinPool的commonPool相关参数配置
  • java.util.concurrent包讲解
    • 线程安全AtomicInteger的讲解
    • CompletableFuture讲解
      • CompletableFuture接口详解
      • CompletableFuture与parallelStream()性能差异
      • CompletableFuture接口详解2
  • Java线程安全
    • 性能与可伸缩性
    • 解决死锁
    • 死锁定义
    • 如何让多线程下的类安全
    • 类的线程安全性定义
    • 实战:实现一个线程安全的单例模式
  • Java常用并发开发工具和类的源码分析
    • CountDownLatch
    • CyclicBarrier
    • Semaphore
    • Exchange
    • ConcurrentHashMap
    • ConcurrentSkipListMap
    • HashMap
      • HashMap源码实现及分析
      • HashMap的一些面试题
    • List
  • Java中的锁
    • 基础知识
    • 番外篇
    • synchronized 是可重入锁吗?为什么?
    • 自旋锁
  • Java多线程的常见问题
    • 常见问题一
Powered by GitBook
On this page
  • 应用场景
  • 案例
  • CompletionService源码分析

Was this helpful?

  1. Java线程池与框架

CompletionService讲解

PreviousFutureTask详解NextFuture、FutureTask、CompletionService、CompletableFuture区别

Last updated 5 years ago

Was this helpful?

应用场景

当向Executor提交多个任务并且希望获得它们在完成之后的结果,如果用FutureTask,可以循环获取task,并调用get方法去获取task执行结果,但是如果task还未完成,获取结果的线程将阻塞直到task完成,由于不知道哪个task优先执行完毕,使用这种方式效率不会很高。在jdk5时候提出接口CompletionService,它整合了Executor和BlockingQueue的功能,可以更加方便在多个任务执行时获取到任务执行结果。

案例

  • 需求:不使用求和公式,计算从1到100000000相加的和。

  • 分析设计:需求指明不能使用求和公式,只能循环依次相加,为了提高效率,我们可以将1到100000000的数分为n段由n个task执行,执行结束后merge结果求最后的和。

  • 代码实现:

    • 声明task执行载体,线程池executor;

    • 声明CompletionService,通过参数指定执行task的线程池,存放已完成状态task的阻塞队列,队列默认为基于链表结构的阻塞队列LinkedBlockingQueue;

    • 调用submit方法提交task;

    • 调用take方法获取已完成状态task。

CompletionService源码分析

CompletionService接口提供五个方法:

  • Future<V> submit(Callable<V> task) 提交Callable类型的task;

  • Future<V> submit(Runnable task, V result) 提交Runnable类型的task;

  • Future<V> take() throws InterruptedException 获取并移除已完成状态的task,如果目前不存在这样的task,则等待;

  • Future<V> poll() 获取并移除已完成状态的task,如果目前不存在这样的task,返回null;

  • Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException 获取并移除已完成状态的task,如果在指定等待时间内不存在这样的task,返回null。

接下来我们来看看CompletionService接口的具体实现:ExecutorCompletionService。

ExecutorCompletionService实现分析

    • executor:执行task的线程池,创建CompletionService必须指定;

    • aes:主要用于创建待执行task;

    • completionQueue:存储已完成状态的task,默认是基于链表结构的阻塞队列LinkedBlockingQueue。

    1. 参数校验,不符合条件的task抛出异常,程序结束;

    2. 将Callable类型或者Runnable类型的task构造成FutureTask;

    3. 把构造好的FutureTask交由线程池executor执行。

    看到这里可能大家会比较疑惑了,task调用submit方法可以提交,完成的task是什么时候被加入到completionQueue里的呢?

    • take在获取并移除已完成状态的task时,如果目前暂时不存在这样的task,等待,直到存在这样的task;

    • poll在获取并移除已完成状态的task时,如果目前暂时不存在这样的task,不等待,直接返回null。

成员变量 ExecutorCompletionService有三个成员变量:

构造方法 ExecutorCompletionService提供两个构造方法,具体的使用具体情况具体分析,使用者可以根据业务场景来进行选择。

task提交 ExecutorCompletionService提供submit方法来提交Callable类型或者Runnable类型的task: 具体的执行流程如下:

针对这个问题,从submit方法的源码可以看出,在提交到线程池的时候需要将FutureTask封装成QueueingFuture,我们来看看QueueingFuture的具体实现: 从源码可以看出,QueueingFuture是FutureTask的子类,实现了done方法,在task执行完成之后将当前task添加到completionQueue,done方法的具体调用在FutureTask的finishCompletion方法,上篇介绍FutureTask的文章已经做过具体的分析,在这里就不再赘述了。

已完成状态task获取 CompletionService的take方法和poll方法都可以获取已完成状态的task,我们来看看具体的实现: 从源码可以看出,take和poll都是调用BlockingQueue提供的方法。既然take和poll都可以获取到已完成状态的task,那么他们的区别是什么呢?

QueueingFuture实现
构造方法
成员变量
CompletionService使用案例
take、poll实现
线程提交.png