ThreadPoolExecutor使用 + 工作机理 + 生命周期

叁叁肆2018-12-18 13:28

此文已由作者赵计刚授权网易云社区发布。

欢迎访问网易云社区,了解更多网易技术产品运营经验。


1、最基础的线程池ThreadPoolExecutor

使用方式:

 1 /**
 2  * ThreadPoolExecutor测试类
 3  * 注意:
 4  * 1、ThreadPoolExecutor是一个线程池
 5  * 2、多个任务都可以由该线程池中选出几条线程来执行
 6  */
 7 public class ThreadPoolExecutorTest {
 8     private static ThreadPoolExecutor executor = 
 9             new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
10     
11     public void executeTask(){
12         Task1 task1 = new Task1();//构建任务1
13         Task2 task2 = new Task2();//构建任务2
14         executor.execute(task1);//执行任务1
15         executor.execute(task2);//执行任务2
16     }
17     
18     /*
19      * 基本任务2
20      */
21     class Task1 implements Runnable{
22         public void run() {
23             //具体任务的业务
24             for(int i=0;i<1000;i++){
25                 System.out.println("hello xxx!!!");
26             }
27         }
28     }
29     
30     /*
31      * 基本任务2
32      */
33     class Task2 implements Runnable{
34         public void run() {
35             //具体任务的业务
36             for(int i=0;i<5;i++){
37                 System.out.println("hello world2!!!");
38             }
39         }
40     }
41     
42     public static void main(String[] args) {
43         ThreadPoolExecutorTest test = new ThreadPoolExecutorTest();
44         test.executeTask();
45     }
46 }

说明:

在代码中,构建了一个线程池(executor)和两个实现了Runnable接口的任务(task1、task2),并将这两个任务提交到executor中去执行。

线程池的配置:集合下边的工作机理与参数详细说明来说。

当然,上述的执行结果是交叉着的,因为存在线程的切换。

 

2、工作机理

A、当一个新的任务被提交到ThreadPoolExecutor的execute()方法中时,如果当前池中正在运行的线程少于corePoolSize,则会创建一个新的线程来处理该任务;

注意:这是池中正在运行的线程,为什么这样说呢?是因为核心线程是每来一个任务才创建一个线程,这个看第三部分。看完第三部分之后,你就会觉得,其实换个说法:"如果当前池中的线程少于corePoolSize"这样会更准确,因为我们也许会通过下边介绍的方法提前将核心线程创建好,如果假设这时候来了一个任务,而所有的核心线程都处于空闲状态的话,这时候是不会去创建新线程的。

B、如果当前池中的线程大于等于corePoolSize,但是小于maximumPoolSize时,如果队列满了,会创建新的线程来处理任务,如果队列没有满,任务加入到队列中去;

C、如果队列满了,正在运行的线程数已经等于maximumPoolSize时,该任务就会被rejected(回绝)

 

3、参数详细说明

A、corePoolSize与maximumPoolSize

  • 如果corePoolSize==maximumPoolSize,线程池的size就是固定的了(这一块儿类似于堆内存的指定,防止扩张带来的损耗,但要视情况而定);
  • 默认情况下,只有当一个新的任务到达时,才会创建和启动core threads,但是可以通过prestartCoreThread和prestartAllCoreThreads来改变;

B、ThreadFactory

  • 通过使用java.util.concurrent.ThreadFactory可以创建新的线程
  • 如果不额外指定ThreadFactory,则使用默认的Executors#defaultThreadFactory
  • 通过该默认的线程工厂,所有创建的线程都会被加入到同一个ThreadGroup中去,并且这些线程都会有相同的优先级(NORM_PRIORITY),并且都是non-daemon线程

注意:这一块儿有一个后台(daemon)线程的概念,典型的后台线程:垃圾回收线程;这个线程与其他应用线程的不同之处在于:当所有的应用线程都没有后,后台线程也就自动消失了。

C、keepAliveTime

  • 如果pool当前拥有的线程超过了corePoolSize,超出的线程如果在大于keepAliveTime的时间外闲置(idle),这些线程就会被终止;
  • 该机制在pool没有被活跃的使用的时候,可以减少资源浪费;
  • 默认情况下,keep-alive机制仅仅会在线程数超过corePoolSizeThreads时才会被使用;
  • 当然,通过使用ThreadPoolExecutor#allowCoreThreadTimeOut(boolean)也可以将这种keep-alive机制应用在core threads上去(只要keepAliveTime>0即可)

D、Queue

任何一种BlockingQueue都可以被用来传递和存储提交到线程池中的任务,有三种队列策略:

1)SynchronousQueue(默认)

  • 直接将任务移交给线程而不是入队,如果已经没有线程立即来处理提交到pool中的任务时,会创建一个新的线程来处理该任务;
  • 这种策略需要maximumPoolSizes无界来确保新提交的任务不会被rejection;
  • 这种方式的最大缺点:当任务到来的速度大于任务被处理的速度时,线程数会疯长。

2)无界队列LinkedBlockingQueue:

  • 由于队列无界,当运行的线程等于corePoolSize时,新到来的任务会入队而不会创建新的线程来执行(即pool中的线程数永远不会大于corePoolSize);
  • 这种方式的缺点:当任务到来的速度大于任务被处理的速度时,队列长度会疯长。

3)有界队列ArrayBlockingQueue:

  • 这种方式是非常难处理好的一种方式,要考虑好ArrayBlockingQueue的大小和maximumPoolSize的大小;
  • 当ArrayBlockingQueue较大而maximumPoolSize较小时,会降低CPU使用率、减少OS资源、减少上下文切换,但是吞吐量会降低。-->线程较少的特点就是这样;
  • 如果任务频繁的被阻塞(例如,they are I/O bound),就需要更多的线程了;
  • 当ArrayBlockingQueue较小而maximumPoolSize较大时,会使CPU使用繁忙但也会遇到一些不可接受的scheduling,吞吐量也会降低。

说明:这一块儿配置是一个比较麻烦的地方,后边会说。

E、回绝任务

执行回绝的场景:看开头部分的工作机理。

在回绝任务的时候,execute()方法会调用RejectedExecutionHandler#rejectedExecution。有四种handler策略:

1)ThreadPoolExecutor.CallerRunsPolicy:调用execute()的线程自己来处理该任务,绝大部分情况下是主线程。

注意:由于主线程执行这个任务,那么新到来的任务就不会被提交到线程池中执行(而是提交到TCP层的队列,TCP层队列满了,就开始拒绝,此时性能已经很低了),直到主线程执行完这个任务。

2)ThreadPoolExecutor.DiscardPolicy:不能被执行的任务会直接被扔掉

3)ThreadPoolExecutor.DiscardOldestPolicy:如果executor没有被关闭,队列头部的任务将会被丢弃,然后将该任务加到队尾

4)ThreadPoolExecutor.AbortPolicy(默认):回绝任务并抛出异常

F、AOP

ThreadPoolExecutor提供了两个方法在每个任务的执行前后进行调用ThreadPoolExecutor#beforeExecute和ThreadPoolExecutor#afterExecute.

 

4、开头实例套用

实例中构建的线程池参数:

  • corePoolSize==5    
  • maximumPoolSize==10  
  • keepAliveTime==30s 
  • 队列:ArrayBlockingQueue,大小为10
  • 线程工厂:defaultThreadFactory(默认)
  • 回绝策略:AbortPolicy(默认)

套一下工作机理:

1)当并发提交了<=5个任务到executor中时(此时任务数<=corePoolSize),executor会使用5个核心线程去执行这些任务;

2)当这时候马上又来了一个任务,如果此时5个核心线程有空闲线程的话,就是用空闲的线程去处理,如果都在忙,这时候该任务进入队列;

3)之后再来任务,还是像第二步那样去执行,直到任务将队列放满了,这时候,如果再来一个任务,如果5个核心线程有空闲线程,直接去执行该任务,如果5个核心线程都在忙,这时候就创建一个新的线程来执行该任务;

4)如果通过上边的流程,最后5个线程都在忙,并且队列满了,并且pool中的线程数已经是10个了(池中的线程总数==maximumPoolSize了),这时候就要执行回绝策略了,在这里,使用了默认的AbortPolicy,即直接放弃该任务并抛出异常。

在代码的执行过程中,如果发现后来创建的5个线程有超过30秒都没被调用过的,该线程就被回收掉了。

 

5、线程池生命周期 

  • 创建之初,状态为RUNNNG
  • 调用了ExecutorService#shutdown:将之前已经提交上来的任务进行处理(包括队列中的),但是不再接收新任务(使用回绝策略回绝新任务),状态SHUNTDOWN
  • 调用了ExecutorService#shutdownNow:取消所有运行中的任务(包括队列中的),并且不再接收新任务(使用回绝策略回绝新任务),状态STOP/TERMINATED

疑问:(这个疑问我会在看完ThreadPoolExecutor的相关源码后进行回答)

当队列满了之后,这时候来了一个任务,恰好5个核心线程有一个空闲了,那么下面两种情况哪一个正确:

1)这个空闲的核心线程直接执行刚刚到来的任务

2)这个空闲的核心线程直接执行队列头部的任务,而将刚刚到来的任务放入队尾

解答:这个问题的答案就一句话,有空闲核心线程,就是用核心线程去执行任务;没有空闲的核心线程,任务才会入队。所以选1)

最后,这里列出上边提到的两种队列的源码解析地址:

ArrayBlockingQueue:第八章 ArrayBlockingQueue源码解析  

LinkedBlockingQueue:第九章 LinkedBlockingQueue源码解析

 

附:线程的生命周期(《实战java高并发程序设计》)




免费领取验证码、内容安全、短信发送、直播点播体验包及云服务器等套餐

更多网易技术、产品、运营经验分享请点击