java 线程池 ThreadPoolExecutor

 

1、简介

 

线程池类为 Java.util.concurrent.ThreadPoolExecutor，经常使用构造方法为：

 

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,

long keepAliveTime, TimeUnit unit,

BlockingQueue<Runnable> workQueue,

RejectedExecutionHandler handler)

 

 

corePoolSize： 线程池维护线程的最少数量

maximumPoolSize：线程池维护线程的最大数量

keepAliveTime： 线程池维护线程所容许的空闲时间

unit： 线程池维护线程所容许的空闲时间的单位

workQueue： 线程池所使用的缓冲队列

handler： 线程池对拒绝任务的处理策略

 

一个任务经过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是Runnable类型对象的run()方法。

 

先看一副图，描述了ThreadPoolExecutor的工做机制: 

 

整个ThreadPoolExecutor的任务处理有4步操做：

 

• 第一步，初始的poolSize < corePoolSize，提交的runnable任务，会直接作为new一个Thread的参数，立马执行
• 第二步，当提交的任务数超过了corePoolSize，就进入了第二步操做。会将当前的runable提交到一个block queue中
• 第三步，若是block queue是个有界队列，当队列满了以后就进入了第三步。若是poolSize < maximumPoolsize时，会尝试new 一个Thread的进行救急处理，立马执行对应的runnable任务
• 第四步，若是第三步救急方案也没法处理了，就会走到第四步执行reject操做。

几点说明：(相信这些网上一搜一大把，我这里简单介绍下，为后面作一下铺垫)

• block queue有如下几种实现：
  1. ArrayBlockingQueue :  有界的数组队列
  2. LinkedBlockingQueue : 可支持有界/无界的队列，使用链表实现
  3. PriorityBlockingQueue : 优先队列，能够针对任务排序
  4. SynchronousQueue : 队列长度为1的队列，和Array有点区别就是：client thread提交到block queue会是一个阻塞过程，直到有一个worker thread链接上来poll task。
• RejectExecutionHandler是针对任务没法处理时的一些自保护处理：
  1. Reject 直接抛出Reject exception
  2. Discard 直接忽略该runnable，不可取
  3. DiscardOldest 丢弃最先入队列的的任务
  4. CallsRun 直接让原先的client thread作为worker线程，进行执行

 

容易被人忽略的点：

1.  pool threads启动后，之后的任务获取都会经过block queue中，获取堆积的runnable task.

 

因此建议： block size >= corePoolSize ，否则线程池就没任何意义

2.  corePoolSize 和 maximumPoolSize的区别， 和你们正常理解的数据库链接池不太同样。

  *  据dbcp pool为例，会有minIdle , maxActive配置。minIdle表明是常驻内存中的threads数量，maxActive表明是工做的最大线程数。

  *  这里的corePoolSize就是链接池的maxActive的概念，它没有minIdle的概念(每一个线程能够设置keepAliveTime，超过多少时间多有任务后销毁线程，但不会固定保持必定数量的threads)。 

  * 这里的maximumPoolSize，是一种救急措施的第一层。当threadPoolExecutor的工做threads存在满负荷，而且block queue队列也满了，这时表明接近崩溃边缘。这时容许临时起一批threads，用来处理runnable，处理完后立马退出。

 

因此建议：  maximumPoolSize >= corePoolSize =指望的最大线程数。 (我曾经配置了corePoolSize=1, maximumPoolSize=20, blockqueue为无界队列，最后就成了单线程工做的pool。典型的配置错误)

 

3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler，从字面意思就是让调用者本身来运行。

咱们常常会在线上使用一些线程池作异步处理，好比我前面作的(业务层)异步并行加载技术分析和设计, 将本来串行的请求都变为了并行操做，但过多的并行会增长系统的负载(好比软中断，上下文切换)。因此确定须要对线程池作一个size限制。可是为了引入异步操做后，避免因在block queue的等待时间过长，因此须要在队列满的时，执行一个callsRun的策略，并行的操做又转为一个串行处理，这样就能够保证尽可能少的延迟影响。

 

因此建议：  RejectExecutionHandler = CallsRun ,  blockqueue size = 2 * poolSize (为啥是2倍poolSize，主要一个考虑就是瞬间高峰处理，容许一个thread等待一个runnable任务)

当一个任务经过execute(Runnable)方法欲添加到线程池时：

 

l  若是此时线程池中的数量小于corePoolSize，即便线程池中的线程都处于空闲状态，也要建立新的线程来处理被添加的任务。

l  若是此时线程池中的数量等于 corePoolSize，可是缓冲队列 workQueue未满，那么任务被放入缓冲队列。

l  若是此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，而且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

l  若是此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，而且线程池中的数量等于maximumPoolSize，那么经过 handler所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，若是三者都满了，使用handler处理被拒绝的任务。

l  当线程池中的线程数量大于 corePoolSize时，若是某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池能够动态的调整池中的线程数。

 

unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：

NANOSECONDS、

MICROSECONDS、

MILLISECONDS、

SECONDS。

 

workQueue经常使用的是：java.util.concurrent.ArrayBlockingQueue

 

handler有四个选择：

ThreadPoolExecutor.AbortPolicy()

抛出java.util.concurrent.RejectedExecutionException异常

 

ThreadPoolExecutor.CallerRunsPolicy()

当抛出RejectedExecutionException异常时，会调用rejectedExecution方法

(若是主线程没有关闭，则主线程调用run方法,源码以下

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }

)

 

ThreadPoolExecutor.DiscardOldestPolicy()

抛弃旧的任务

 

ThreadPoolExecutor.DiscardPolicy()

抛弃当前的任务

 

2、相关参考

 

一个 ExecutorService，它使用可能的几个池线程之一执行每一个提交的任务，一般使用 Executors 工厂方法配置。

 

线程池能够解决两个不一样问题：因为减小了每一个任务调用的开销，它们一般能够在执行大量异步任务时提供加强的性能，而且还能够提供绑定和管理资源（包括执行集合任务时使用的线程）的方法。每一个ThreadPoolExecutor 还维护着一些基本的统计数据，如完成的任务数。

 

为了便于跨大量上下文使用，此类提供了不少可调整的参数和扩展挂钩。可是，强烈建议程序员使用较为方便的 Executors 工厂方法 Executors.newCachedThreadPool()（无界线程池，能够进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池）和 Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预约义了设置。不然，在手动配置和调整此类时，使用如下指导：

 

核心和最大池大小

ThreadPoolExecutor 将根据 corePoolSize（参见 getCorePoolSize()）和 maximumPoolSize（参见getMaximumPoolSize()）设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时，若是运行的线程少于 corePoolSize，则建立新线程来处理请求，即便其余辅助线程是空闲的。若是运行的线程多于corePoolSize 而少于 maximumPoolSize，则仅当队列满时才建立新线程。若是设置的 corePoolSize 和 maximumPoolSize相同，则建立了固定大小的线程池。若是将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则容许池适应任意数量的并发任务。在大多数状况下，核心和最大池大小仅基于构造来设置，不过也可使用setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改