java并发编程之ThreadPoolExecutor

短命女 2021-09-17 11:30 384阅读 0赞

#     #

包 java.util.concurrent.\*  
\[ 一 \]、常用线程池

最常用构造方法为：

ThreadPoolExecutor(int corePoolSize,  
                   int maximumPoolSize,  
                   long keepAliveTime,  
                   TimeUnit unit,  
                   BlockingQueue<Runnable> workQueue,

ThreadFactory threadFactory,

RejectedExecutionHandler handler)

JDK自带的配置好的线程池：

// 固定工作线程数量的线程池  
        ExecutorService executorService1 = Executors.newFixedThreadPool(3);  
  
        // 一个可缓存的线程池  
        ExecutorService executorService2 = Executors.newCachedThreadPool();  
  
        // 单线程化的Executor  
        ExecutorService executorService3 = Executors.newSingleThreadExecutor();  
  
        // 支持定时的以及周期性的任务执行  
        ExecutorService executorService4 = Executors.newScheduledThreadPool(3);  
这些预定义好的线程池服务也是基于ThreadPoolExecutor配置的，所以我们应该从最基本的参数着手了解，如下：  
  
参数详细说明：  
\[ 1 \]、corePoolSize： 线程池维护线程的最少数量  
\[ 2 \]、maximumPoolSize：线程池维护线程的最大数量  
\[ 3 \]、keepAliveTime： 线程池维护线程所允许的空闲时间  
\[ 4 \]、unit： 线程池维护线程所允许的空闲时间的单位，unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：

*  NANOSECONDS
 *  MICROSECONDS
 *  MILLISECONDS
 *  SECONDS

\[ 5\]、 workQueue： 线程池所使用的缓冲队列，常用的是：java.util.concurrent.ArrayBlockingQueue

\[6 \]    threadFactory 负责给线程池创建线程的工厂。线程池中的线程是由ThreadFactory创建。如果不特别指定，会使用Executors.defaultThreadFactory创建位于同一个线程组，相同优先级（NORM\_PRIORITY）的非守护线程。如果由你来指定ThreadFactory，你可以定制线程名字，线程组，优先级，是否为守护线程等属性。  
\[ 7 \]、 handler： 线程池对拒绝任务的处理策略，有四个选择如下：

*  ThreadPoolExecutor.AbortPolicy()：抛出java.util.concurrent.RejectedExecutionException异常
 *  ThreadPoolExecutor.CallerRunsPolicy()：重试添加当前的任务，他会自动重复调用execute()方法
 *  ThreadPoolExecutor.DiscardOldestPolicy()：抛弃旧的任务
 *  ThreadPoolExecutor.DiscardPolicy()：抛弃当前的任务

\[ 二 \]、详细说明   
\[ 1 \]、当一个任务通过execute(Runnable)方法欲添加到线程池时：

*   *  如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。
     *  如果此时线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。
     *  如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。
     *  如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于 maximumPoolSize，那么通过 handler所指定的策略来处理此任务。也就是：处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程 maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。
     *  当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

ThreadPoolExecutor的工作机制:

![点击查看原始大小图片][d9c97d76-aa58-3e7d-9f43-33bf90c1d4e8.jpg]

整个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，从字面意思就是让调用者自己来运行。

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

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

参考:[http://www.iteye.com/topic/1118660][http_www.iteye.com_topic_1118660]

http://blog.csdn.net/waterbig/article/details/4794244

http://rdc.taobao.com/team/jm/archives/595

[d9c97d76-aa58-3e7d-9f43-33bf90c1d4e8.jpg]: /images/20210726/fa10f87f08c44ce8a9d978df233c9222.png
[Link 1]: http://www.iteye.com/blog/918898
[http_www.iteye.com_topic_1118660]: http://www.iteye.com/topic/1118660