ThreadPoolExecutor的理解思考

工作中多处接触到了ThreadPoolExecutor。趁着现在还算空,学习总结一下。

前记:

  1. jdk官方文档(javadoc)是学习的最好,最权威的参考。
  2. 文章分上中下。上篇中主要介绍ThreadPoolExecutor接受任务相关的两方面入参的意义和区别,池大小参数corePoolSize和maximumPoolSize,BlockingQueue选型(SynchronousQueue,LinkedBlockingQueue,ArrayBlockingQueue);中篇中主要聊聊与keepAliveTime这个参数相关的话题;下片中介绍一下一些比较少用的该类的API,及他的近亲:ScheduledThreadPoolExecutor
  3. 如果理解错误,请直接指出。

查看JDK帮助文档,可以发现该类比较简单,继承自AbstractExecutorService,而AbstractExecutorService实现了ExecutorService接口。

ThreadPoolExecutor的完整构造方法的签名是:

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

先记着,后面慢慢解释。

===============================神奇分割线==================================

其实对于ThreadPoolExecutor的构造函数网上有N多的解释的,大多讲得都很好,不过我想先换个方式,从Executors这个类入手。因为他的几个构造工厂构造方法名字取得令人很容易了解有什么特点。但是其实Executors类的底层实现便是ThreadPoolExecutor!

ThreadPoolExecutor是Executors类的底层实现。

在JDK帮助文档中,有如此一段话:

“强烈建议程序员使用较为方便的 Executors 工厂方法 Executors.newCachedThreadPool()(无界线程池,可以进行自动线程回收)、Executors.newFixedThreadPool(int)(固定大小线程池)和Executors.newSingleThreadExecutor()(单个后台线程),它们均为大多数使用场景预定义了设置。”

可以推断出ThreadPoolExecutor与Executors类必然关系密切。

===============================神奇分割线==================================

OK,那就来看看源码吧,从newFixedThreadPool开始。

ExecutorService newFixedThreadPool(int nThreads):固定大小线程池。

可以看到,corePoolSize和maximumPoolSize的大小是一样的(实际上,后面会介绍,如果使用无界queue的话maximumPoolSize参数是没有意义的),keepAliveTime和unit的设值表名什么?-就是该实现不想keep alive!最后的BlockingQueue选择了LinkedBlockingQueue,该queue有一个特点,他是无界的。

Java代码  

  1. public static ExecutorService newFixedThreadPool(int nThreads) {
  2. return new ThreadPoolExecutor(nThreads, nThreads,
  3. 0L, TimeUnit.MILLISECONDS,
  4. new LinkedBlockingQueue<Runnable>());
  5. }

ExecutorService newSingleThreadExecutor():单线程。

可以看到,与fixedThreadPool很像,只不过fixedThreadPool中的入参直接退化为1

Java代码  

  1. public static ExecutorService newSingleThreadExecutor() {
  2. return new FinalizableDelegatedExecutorService
  3. (new ThreadPoolExecutor(1, 1,
  4. 0L, TimeUnit.MILLISECONDS,
  5. new LinkedBlockingQueue<Runnable>()));
  6. }

ExecutorService newCachedThreadPool():无界线程池,可以进行自动线程回收。

这个实现就有意思了。首先是无界的线程池,所以我们可以发现maximumPoolSize为big big。其次BlockingQueue的选择上使用SynchronousQueue。可能对于该BlockingQueue有些陌生,简单说:该QUEUE中,每个插入操作必须等待另一个

线程的对应移除操作。比如,我先添加一个元素,接下来如果继续想尝试添加则会阻塞,直到另一个线程取走一个元素,反之亦然。(想到什么?就是缓冲区为1的生产者消费者模式^_^)

注意到介绍中的自动回收线程的特性吗,为什么呢?先不说,但注意到该实现中corePoolSize和maximumPoolSize的大小不同。

Java代码  

  1. public static ExecutorService newCachedThreadPool() {
  2. return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
  3. 60L, TimeUnit.SECONDS,
  4. new SynchronousQueue<Runnable>());
  5. }

===============================神奇分割线==================================

到此如果有很多疑问,那是必然了(除非你也很了解了)

先从BlockingQueue<Runnable> workQueue这个入参开始说起。在JDK中,其实已经说得很清楚了,一共有三种类型的queue。以下为引用:(我会稍微修改一下,并用红色突出显示)

所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互:

  • 如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。(什么意思?如果当前运行的线程小于corePoolSize,则任务根本不会存放,添加到queue中,而是直接抄家伙(thread)开始运行)
  • 如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程
  • 如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。

先不着急举例子,因为首先需要知道queue上的三种类型。

排队有三种通用策略:

  1. 直接提交。工作队列的默认选项是 SynchronousQueue,它将任务直接提交给线程而不保持它们。在此,如果不存在可用于立即运行任务的线程,则试图把任务加入队列将失败,因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。
  2. 无界队列。使用无界队列(例如,不具有预定义容量的 LinkedBlockingQueue)将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样,创建的线程就不会超过 corePoolSize。(因此,maximumPoolSize 的值也就无效了。)当每个任务完全独立于其他任务,即任务执行互不影响时,适合于使用无界队列;例如,在 Web 页服务器中。这种排队可用于处理瞬态突发请求,当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。
  3. 有界队列。当使用有限的 maximumPoolSizes 时,有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽,但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷:使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销,但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如,如果它们是 I/O 边界),则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小,CPU 使用率较高,但是可能遇到不可接受的调度开销,这样也会降低吞吐量。

===============================神奇分割线==================================

到这里,该了解的理论已经够多了,可以调节的就是corePoolSize和maximumPoolSizes 这对参数还有就是BlockingQueue的选择。

例子一:使用直接提交策略,也即SynchronousQueue。

首先SynchronousQueue是无界的,也就是说他存数任务的能力是没有限制的,但是由于该Queue本身的特性,在某次添加元素后必须等待其他线程取走后才能继续添加。在这里不是核心线程便是新创建的线程,但是我们试想一样下,下面的场景。

我们使用一下参数构造ThreadPoolExecutor:

Java代码  

  1. new ThreadPoolExecutor(
  2. 2, 3, 30, TimeUnit.SECONDS,
  3. new SynchronousQueue<Runnable>(),
  4. new RecorderThreadFactory("CookieRecorderPool"),
  5. new ThreadPoolExecutor.CallerRunsPolicy());

当核心线程已经有2个正在运行.

  1. 此时继续来了一个任务(A),根据前面介绍的“如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。”,所以A被添加到queue中。
  2. 又来了一个任务(B),且核心2个线程还没有忙完,OK,接下来首先尝试1中描述,但是由于使用的SynchronousQueue,所以一定无法加入进去。
  3. 此时便满足了上面提到的“如果无法将请求加入队列,则创建新的线程,除非创建此线程超出maximumPoolSize,在这种情况下,任务将被拒绝。”,所以必然会新建一个线程来运行这个任务。
  4. 暂时还可以,但是如果这三个任务都还没完成,连续来了两个任务,第一个添加入queue中,后一个呢?queue中无法插入,而线程数达到了maximumPoolSize,所以只好执行异常策略了。

所以在使用SynchronousQueue通常要求maximumPoolSize是无界的,这样就可以避免上述情况发生(如果希望限制就直接使用有界队列)。对于使用SynchronousQueue的作用jdk中写的很清楚:此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。

什么意思?如果你的任务A1,A2有内部关联,A1需要先运行,那么先提交A1,再提交A2,当使用SynchronousQueue我们可以保证,A1必定先被执行,在A1么有被执行前,A2不可能添加入queue中

例子二:使用无界队列策略,即LinkedBlockingQueue

这个就拿newFixedThreadPool来说,根据前文提到的规则:

写道

如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。

那么当任务继续增加,会发生什么呢?

写道

如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。

OK,此时任务变加入队列之中了,那什么时候才会添加新线程呢?

写道

如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。

这里就很有意思了,可能会出现无法加入队列吗?不像SynchronousQueue那样有其自身的特点,对于无界队列来说,总是可以加入的(资源耗尽,当然另当别论)。换句说,永远也不会触发产生新的线程!corePoolSize大小的线程数会一直运行,忙完当前的,就从队列中拿任务开始运行。所以要防止任务疯长,比如任务运行的实行比较长,而添加任务的速度远远超过处理任务的时间,而且还不断增加,如果任务内存大一些,不一会儿就爆了,呵呵。

可以仔细想想哈。

例子三:有界队列,使用ArrayBlockingQueue。

这个是最为复杂的使用,所以JDK不推荐使用也有些道理。与上面的相比,最大的特点便是可以防止资源耗尽的情况发生。

举例来说,请看如下构造方法:

Java代码  

  1. new ThreadPoolExecutor(
  2. 2, 4, 30, TimeUnit.SECONDS,
  3. new ArrayBlockingQueue<Runnable>(2),
  4. new RecorderThreadFactory("CookieRecorderPool"),
  5. new ThreadPoolExecutor.CallerRunsPolicy());

假设,所有的任务都永远无法执行完。

对于首先来的A,B来说直接运行,接下来,如果来了C,D,他们会被放到queu中,如果接下来再来E,F,则增加线程运行E,F。但是如果再来任务,队列无法再接受了,线程数也到达最大的限制了,所以就会使用拒绝策略来处理。

总结:

  1. ThreadPoolExecutor的使用还是很有技巧的。
  2. 使用无界queue可能会耗尽系统资源。
  3. 使用有界queue可能不能很好的满足性能,需要调节线程数和queue大小
  4. 线程数自然也有开销,所以需要根据不同应用进行调节。

通常来说对于静态任务可以归为:

  1. 数量大,但是执行时间很短
  2. 数量小,但是执行时间较长
  3. 数量又大执行时间又长
  4. 除了以上特点外,任务间还有些内在关系

看完这篇问文章后,希望能够可以选择合适的类型了

原文地址:http://dongxuan.iteye.com/blog/901689

时间: 2024-09-27 04:35:13

ThreadPoolExecutor的理解思考的相关文章

ThreadPoolExecutor使用和思考(上)-线程池大小设置与BlockingQueue的三种实现区别

工作中多处接触到了ThreadPoolExecutor.趁着现在还算空,学习总结一下. 前记: jdk官方文档(javadoc)是学习的最好,最权威的参考. 文章分上中下.上篇中主要介绍ThreadPoolExecutor接受任务相关的两方面入参的意义和区别,池大小参数corePoolSize和maximumPoolSize,BlockingQueue选型(SynchronousQueue,LinkedBlockingQueue,ArrayBlockingQueue):中篇中主要聊聊与keepA

ThreadPoolExecutor使用和思考-线程池大小设置与BlockingQueue的三种

工作中多处接触到了ThreadPoolExecutor.趁着现在还算空,学习总结一下. 前记: jdk官方文档(javadoc)是学习的最好,最权威的参考. 文章分上中下.上篇中主要介绍ThreadPoolExecutor接受任务相关的两方面入参的意义和区别,池大小参数corePoolSize和maximumPoolSize,BlockingQueue选型(SynchronousQueue,LinkedBlockingQueue,ArrayBlockingQueue):中篇中主要聊聊与keepA

LoadRunner 技巧之 思考时间设置

用户访问某个网站或软件,一般不会不停地做个各种操作,例如一次查询,用户需要时间查看查询的结果是否是自己想要的.例如一次订单提交,用户需要时间核对自己填写的信息是否正确等. 也就是说用户在做某些操作时,是会有停留时间的,我把这个时间叫思考时间.但利用代码去执行的时候是没有时间的,当然,脚本运行本身是需要时间的,但比起人的思考时间要小很多.这也是我们为什么要用软件来代替人的某些工作. 但有时候,我们在做性能测试时,为了更真实的模拟用户的操作,需要给代码加入思考时间.来看看在loadrunner是如何

Java 线程池学习

Reference: <创建Java线程池>[1],<Java线程:新特征-线程池>[2], <Java线程池学习>[3],<线程池ThreadPoolExecutor使用简介>[4],<Java5中的线程池实例讲解>[5],<ThreadPoolExecutor使用和思考>[6] [1]中博主自己通过ThreadGroup实现一个线程池(挺方便理解的),使用的是jdk1.4版本,Jdk1.5版本以上提供了现成的线程池. [2]中介绍

Android Tips – 填坑手册

出于: androidChina   http://www.androidchina.net/3595.html 学习 Android 至今,大大小小的坑没少踩,庆幸的是,在强大的搜索引擎与无私奉献的人们的帮助下,我遇到的坑都顺利地被填平了. 为了便于日后遇到同样的问题时,能免于再次搜索带来的麻烦,我养成了收藏书签的习惯,随着书签(Tips)的日积月累,我想,是时候该有这个项目了. 如果你是个 Android 新人,那么我希望这份列表,可以成为你踩到坑时的不完全手册. 当然,这份列表一定会有遗漏

Hadoop Day3

MapReduce原理(****理解***) 思考:怎样解决海量数据的计算? MapReduce概述 MapReduce是一种分布式计算模型,由Google提出,主要用于搜索领域,解决海量数据的计算问题. MapReduce由两个阶段组成: map():任务分解 reduce():结果汇总 这两个函数的形参是key.value对,表示函数的输入信息. 数据类型与格式 map.reduce键值对格式: 函数 输入键值对 输出键值对 Map() <K1,V1> <K2,V2> Redu

《深入浅出数据分析》读后具体解释

<深入浅出数据分析>为数据挖掘入门级教材.通篇以一个化妆品公司Acme为典型案例,分析了从数据分析到决策的全过程.本篇为读后总结,增加自己的练习回答.同一时候贴出书中答案.供对照. 第一部分 高效读书 1.人的大脑会拼命阻止普通.常规.一般的事情,以免干扰自己记录重要的事情. 不是每件事情都如同洪水猛兽般突然.紧急.重要,因此虽然理性上你觉得这件事很重要,可是大脑会习惯性地阻止.有两种方法让大脑像吃人的老虎一样记住知识: 一.缓慢而乏味--反复灌输. 二.迅速而有效--做一切促进不同类型大脑活

【转载】虫师『性能测试』文章大汇总

虫师『性能测试』文章大汇总 为了方便阅读,我重新整理本文,将包含本博客所有与性能测试有关的内容. ------------------------------------------- 近两年市面上的性能测试书籍很多了,但大部分书都在讲loadrunner的操作技巧项目与项目实践.我不认为有什么问题,因为loadrunner性能测试工具已经占据很大市场.loadrunner是非常的强大,但我们在做性能测试时,往往都以“loadrunner的模式”在思考如何进行性能测试.loadrunner只是一

《性能测试诊断分析与优化》学习总结(一)

我始终坚信,一本好书,可以改变人的一生,成功一定有方法,失败一定有原因,而我个人认为,失败的最大原因是学习不够.不断学习,活到老,学到老. 作为软件测试人员来说,或多或少都会对性能测试有点感兴趣,我个人认为,软件性能与软件功能的最大兴趣在于从用户体检的角度,感受软件的质量度量,如响应时间.准确性.成功率等. 性能测试的目标是通过制定性能测试策略.性能测试计划.执行性能测试,并对性能问题进行定位分析和优化.影响软件系统的性能主要因素包括:硬件设施.网络.操作系统.并发用户数.数据库数据量.中间件等