Java并发编程:线程池的使用(很是棒,通俗易懂)

2020年01月11日 阅读数:52
这篇文章主要向大家介绍Java并发编程:线程池的使用(很是棒,通俗易懂),主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。

Java并发编程:线程池的使用html

  在前面的文章中,咱们使用线程的时候就去建立一个线程,这样实现起来很是简便,可是就会有一个问题:java

  若是并发的线程数量不少,而且每一个线程都是执行一个时间很短的任务就结束了,这样频繁建立线程就会大大下降系统的效率,由于频繁建立线程和销毁线程须要时间。编程

  那么有没有一种办法使得线程能够复用,就是执行完一个任务,并不被销毁,而是能够继续执行其余的任务?数组

  在Java中能够经过线程池来达到这样的效果。今天咱们就来详细讲解一下Java的线程池,首先咱们从最核心的ThreadPoolExecutor类中的方法讲起,而后再讲述它的实现原理,接着给出了它的使用示例,最后讨论了一下如何合理配置线程池的大小。缓存

  如下是本文的目录大纲:并发

  一.Java中的ThreadPoolExecutor类ide

  二.深刻剖析线程池实现原理ui

  三.使用示例this

  四.如何合理配置线程池的大小 spa

  如有不正之处请多多谅解,并欢迎批评指正。

  请尊重做者劳动成果,转载请标明原文连接:

  http://www.cnblogs.com/dolphin0520/p/3932921.html

 

一.Java中的ThreadPoolExecutor类

  java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类,所以若是要透彻地了解Java中的线程池,必须先了解这个类。下面咱们来看一下ThreadPoolExecutor类的具体实现源码。

  在ThreadPoolExecutor类中提供了四个构造方法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public  class  ThreadPoolExecutor  extends  AbstractExecutorService {
     .....
     public  ThreadPoolExecutor( int  corePoolSize, int  maximumPoolSize, long  keepAliveTime,TimeUnit unit,
             BlockingQueue<Runnable> workQueue);
 
     public  ThreadPoolExecutor( int  corePoolSize, int  maximumPoolSize, long  keepAliveTime,TimeUnit unit,
             BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
     public  ThreadPoolExecutor( int  corePoolSize, int  maximumPoolSize, long  keepAliveTime,TimeUnit unit,
             BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
     public  ThreadPoolExecutor( int  corePoolSize, int  maximumPoolSize, long  keepAliveTime,TimeUnit unit,
         BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
     ...
}

   从上面的代码能够得知,ThreadPoolExecutor继承了AbstractExecutorService类,并提供了四个构造器,事实上,经过观察每一个构造器的源码具体实现,发现前面三个构造器都是调用的第四个构造器进行的初始化工做。

   下面解释下一下构造器中各个参数的含义:

  • corePoolSize:核心池的大小,这个参数跟后面讲述的线程池的实现原理有很是大的关系。在建立了线程池后,默认状况下,线程池中并无任何线程,而是等待有任务到来才建立线程去执行任务,除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法,从这2个方法的名字就能够看出,是预建立线程的意思,即在没有任务到来以前就建立corePoolSize个线程或者一个线程。默认状况下,在建立了线程池后,线程池中的线程数为0,当有任务来以后,就会建立一个线程去执行任务,当线程池中的线程数目达到corePoolSize后,就会把到达的任务放到缓存队列当中;
  • maximumPoolSize:线程池最大线程数,这个参数也是一个很是重要的参数,它表示在线程池中最多能建立多少个线程;
  • keepAliveTime:表示线程没有任务执行时最多保持多久时间会终止。默认状况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起做用,直到线程池中的线程数不大于corePoolSize,即当线程池中的线程数大于corePoolSize时,若是一个线程空闲的时间达到keepAliveTime,则会终止,直到线程池中的线程数不超过corePoolSize。可是若是调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起做用,直到线程池中的线程数为0;
  • unit:参数keepAliveTime的时间单位,有7种取值,在TimeUnit类中有7种静态属性:
复制代码
TimeUnit.DAYS;               //
TimeUnit.HOURS;             //小时
TimeUnit.MINUTES;           //分钟
TimeUnit.SECONDS;           //
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //纳秒
复制代码
  • workQueue:一个阻塞队列,用来存储等待执行的任务,这个参数的选择也很重要,会对线程池的运行过程产生重大影响,通常来讲,这里的阻塞队列有如下几种选择:
ArrayBlockingQueue;
LinkedBlockingQueue;
SynchronousQueue;

  ArrayBlockingQueue和PriorityBlockingQueue使用较少,通常使用LinkedBlockingQueue和Synchronous。线程池的排队策略与BlockingQueue有关。

  • threadFactory:线程工厂,主要用来建立线程;
  • handler:表示当拒绝处理任务时的策略,有如下四种取值:
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 
ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,可是不抛出异常。 
ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,而后从新尝试执行任务(重复此过程)
ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务 

   具体参数的配置与线程池的关系将在下一节讲述。

  从上面给出的ThreadPoolExecutor类的代码能够知道,ThreadPoolExecutor继承了AbstractExecutorService,咱们来看一下AbstractExecutorService的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public  abstract  class  AbstractExecutorService  implements  ExecutorService {
 
     
     protected  <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
     protected  <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
     public  Future<?> submit(Runnable task) {};
     public  <T> Future<T> submit(Runnable task, T result) { };
     public  <T> Future<T> submit(Callable<T> task) { };
     private  <T> T doInvokeAny(Collection<?  extends  Callable<T>> tasks,
                             boolean  timed,  long  nanos)
         throws  InterruptedException, ExecutionException, TimeoutException {
     };
     public  <T> T invokeAny(Collection<?  extends  Callable<T>> tasks)
         throws  InterruptedException, ExecutionException {
     };
     public  <T> T invokeAny(Collection<?  extends  Callable<T>> tasks,
                            long  timeout, TimeUnit unit)
         throws  InterruptedException, ExecutionException, TimeoutException {
     };
     public  <T> List<Future<T>> invokeAll(Collection<?  extends  Callable<T>> tasks)
         throws  InterruptedException {
     };
     public  <T> List<Future<T>> invokeAll(Collection<?  extends  Callable<T>> tasks,
                                          long  timeout, TimeUnit unit)
         throws  InterruptedException {
     };
}

   AbstractExecutorService是一个抽象类,它实现了ExecutorService接口。

  咱们接着看ExecutorService接口的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public  interface  ExecutorService  extends  Executor {
 
     void  shutdown();
     boolean  isShutdown();
     boolean  isTerminated();
     boolean  awaitTermination( long  timeout, TimeUnit unit)
         throws  InterruptedException;
     <T> Future<T> submit(Callable<T> task);
     <T> Future<T> submit(Runnable task, T result);
     Future<?> submit(Runnable task);
     <T> List<Future<T>> invokeAll(Collection<?  extends  Callable<T>> tasks)
         throws  InterruptedException;
     <T> List<Future<T>> invokeAll(Collection<?  extends  Callable<T>> tasks,
                                   long  timeout, TimeUnit unit)
         throws  InterruptedException;
 
     <T> T invokeAny(Collection<?  extends  Callable<T>> tasks)
         throws  InterruptedException, ExecutionException;
     <T> T invokeAny(Collection<?  extends  Callable<T>> tasks,
                     long  timeout, TimeUnit unit)
         throws  InterruptedException, ExecutionException, TimeoutException;
}

   而ExecutorService又是继承了Executor接口,咱们看一下Executor接口的实现:

1
2
3
public  interface  Executor {
     void  execute(Runnable command);
}

   到这里,你们应该明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了。

  Executor是一个顶层接口,在它里面只声明了一个方法execute(Runnable),返回值为void,参数为Runnable类型,从字面意思能够理解,就是用来执行传进去的任务的;

  而后ExecutorService接口继承了Executor接口,并声明了一些方法:submit、invokeAll、invokeAny以及shutDown等;

  抽象类AbstractExecutorService实现了ExecutorService接口,基本实现了ExecutorService中声明的全部方法;

  而后ThreadPoolExecutor继承了类AbstractExecutorService。

  在ThreadPoolExecutor类中有几个很是重要的方法:

1
2
3
4
execute()
submit()
shutdown()
shutdownNow()

   execute()方法其实是Executor中声明的方法,在ThreadPoolExecutor进行了具体的实现,这个方法是ThreadPoolExecutor的核心方法,经过这个方法能够向线程池提交一个任务,交由线程池去执行。

  submit()方法是在ExecutorService中声明的方法,在AbstractExecutorService就已经有了具体的实现,在ThreadPoolExecutor中并无对其进行重写,这个方法也是用来向线程池提交任务的,可是它和execute()方法不一样,它可以返回任务执行的结果,去看submit()方法的实现,会发现它实际上仍是调用的execute()方法,只不过它利用了Future来获取任务执行结果(Future相关内容将在下一篇讲述)。

  shutdown()和shutdownNow()是用来关闭线程池的。

  还有不少其余的方法:

  好比:getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等获取与线程池相关属性的方法,有兴趣的朋友能够自行查阅API。

二.深刻剖析线程池实现原理

  在上一节咱们从宏观上介绍了ThreadPoolExecutor,下面咱们来深刻解析一下线程池的具体实现原理,将从下面几个方面讲解:

  1.线程池状态

  2.任务的执行

  3.线程池中的线程初始化

  4.任务缓存队列及排队策略

  5.任务拒绝策略

  6.线程池的关闭

  7.线程池容量的动态调整

 

1.线程池状态

  在ThreadPoolExecutor中定义了一个volatile变量,另外定义了几个static final变量表示线程池的各个状态:

1
2
3
4
5
volatile  int  runState;
static  final  int  RUNNING    =  0 ;
static  final  int  SHUTDOWN   =  1 ;
static  final  int  STOP       =  2 ;
static  final  int  TERMINATED =  3 ;

   runState表示当前线程池的状态,它是一个volatile变量用来保证线程之间的可见性;

  下面的几个static final变量表示runState可能的几个取值。

  当建立线程池后,初始时,线程池处于RUNNING状态;

  若是调用了shutdown()方法,则线程池处于SHUTDOWN状态,此时线程池不可以接受新的任务,它会等待全部任务执行完毕;

  若是调用了shutdownNow()方法,则线程池处于STOP状态,此时线程池不能接受新的任务,而且会去尝试终止正在执行的任务;

  当线程池处于SHUTDOWN或STOP状态,而且全部工做线程已经销毁,任务缓存队列已经清空或执行结束后,线程池被设置为TERMINATED状态。

2.任务的执行

  在了解将任务提交给线程池到任务执行完毕整个过程以前,咱们先来看一下ThreadPoolExecutor类中其余的一些比较重要成员变量:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private  final  BlockingQueue<Runnable> workQueue;               //任务缓存队列,用来存放等待执行的任务
private  final  ReentrantLock mainLock =  new  ReentrantLock();    //线程池的主要状态锁,对线程池状态(好比线程池大小
                                                               //、runState等)的改变都要使用这个锁
private  final  HashSet<Worker> workers =  new  HashSet<Worker>();   //用来存放工做集
 
private  volatile  long   keepAliveTime;     //线程存货时间   
private  volatile  boolean  allowCoreThreadTimeOut;    //是否容许为核心线程设置存活时间
private  volatile  int    corePoolSize;      //核心池的大小(即线程池中的线程数目大于这个参数时,提交的任务会被放进任务缓存队列)
private  volatile  int    maximumPoolSize;    //线程池最大能容忍的线程数
 
private  volatile  int    poolSize;        //线程池中当前的线程数
 
private  volatile  RejectedExecutionHandler handler;  //任务拒绝策略
 
private  volatile  ThreadFactory threadFactory;    //线程工厂,用来建立线程
 
private  int  largestPoolSize;    //用来记录线程池中曾经出现过的最大线程数
 
private  long  completedTaskCount;    //用来记录已经执行完毕的任务个数

   每一个变量的做用都已经标明出来了,这里要重点解释一下corePoolSize、maximumPoolSize、largestPoolSize三个变量。

  corePoolSize在不少地方被翻译成核心池大小,其实个人理解这个就是线程池的大小。举个简单的例子:

  假若有一个工厂,工厂里面有10个工人,每一个工人同时只能作一件任务。

  所以只要当10个工人中有工人是空闲的,来了任务就分配给空闲的工人作;

  当10个工人都有任务在作时,若是还来了任务,就把任务进行排队等待;

  若是说新任务数目增加的速度远远大于工人作任务的速度,那么此时工厂主管可能会想补救措施,好比从新招4个临时工人进来;

  而后就将任务也分配给这4个临时工人作;

  若是说着14个工人作任务的速度仍是不够,此时工厂主管可能就要考虑再也不接收新的任务或者抛弃前面的一些任务了。

  当这14个工人当中有人空闲时,而新任务增加的速度又比较缓慢,工厂主管可能就考虑辞掉4个临时工了,只保持原来的10个工人,毕竟请额外的工人是要花钱的。

 

  这个例子中的corePoolSize就是10,而maximumPoolSize就是14(10+4)。

  也就是说corePoolSize就是线程池大小,maximumPoolSize在我看来是线程池的一种补救措施,即任务量忽然过大时的一种补救措施。

  不过为了方便理解,在本文后面仍是将corePoolSize翻译成核心池大小。

  largestPoolSize只是一个用来起记录做用的变量,用来记录线程池中曾经有过的最大线程数目,跟线程池的容量没有任何关系。

 

  下面咱们进入正题,看一下任务从提交到最终执行完毕经历了哪些过程。

  在ThreadPoolExecutor类中,最核心的任务提交方法是execute()方法,虽然经过submit也能够提交任务,可是实际上submit方法里面最终调用的仍是execute()方法,因此咱们只须要研究execute()方法的实现原理便可:

1
2
3
4
5
6
7
8
9
10
11
12
public  void  execute(Runnable command) {
     if  (command ==  null )
         throw  new  NullPointerException();
     if  (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
         if  (runState == RUNNING && workQueue.offer(command)) {
             if  (runState != RUNNING || poolSize ==  0 )
                 ensureQueuedTaskHandled(command);
         }
         else  if  (!addIfUnderMaximumPoolSize(command))
             reject(command);  // is shutdown or saturated
     }
}

   上面的代码可能看起来不是那么容易理解,下面咱们一句一句解释:

  首先,判断提交的任务command是否为null,如果null,则抛出空指针异常;

  接着是这句,这句要好好理解一下:

1
if  (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))

   因为是或条件运算符,因此先计算前半部分的值,若是线程池中当前线程数不小于核心池大小,那么就会直接进入下面的if语句块了。

  若是线程池中当前线程数小于核心池大小,则接着执行后半部分,也就是执行

1
addIfUnderCorePoolSize(command)

  若是执行完addIfUnderCorePoolSize这个方法返回false,则继续执行下面的if语句块,不然整个方法就直接执行完毕了。

  若是执行完addIfUnderCorePoolSize这个方法返回false,而后接着判断:

1
if  (runState == RUNNING && workQueue.offer(command))

   若是当前线程池处于RUNNING状态,则将任务放入任务缓存队列;若是当前线程池不处于RUNNING状态或者任务放入缓存队列失败,则执行:

1
addIfUnderMaximumPoolSize(command)

  若是执行addIfUnderMaximumPoolSize方法失败,则执行reject()方法进行任务拒绝处理。

  回到前面:

1
if  (runState == RUNNING && workQueue.offer(command))

   这句的执行,若是说当前线程池处于RUNNING状态且将任务放入任务缓存队列成功,则继续进行判断:

1
if  (runState != RUNNING || poolSize ==  0 )

   这句判断是为了防止在将此任务添加进任务缓存队列的同时其余线程忽然调用shutdown或者shutdownNow方法关闭了线程池的一种应急措施。若是是这样就执行:

1
ensureQueuedTaskHandled(command)

   进行应急处理,从名字能够看出是保证 添加到任务缓存队列中的任务获得处理。

  咱们接着看2个关键方法的实现:addIfUnderCorePoolSize和addIfUnderMaximumPoolSize:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private  boolean  addIfUnderCorePoolSize(Runnable firstTask) {
     Thread t =  null ;
     final  ReentrantLock mainLock =  this .mainLock;
     mainLock.lock();
     try  {
         if  (poolSize < corePoolSize && runState == RUNNING)
             t = addThread(firstTask);         //建立线程去执行firstTask任务   
         finally  {
         mainLock.unlock();
     }
     if  (t ==  null )
         return  false ;
     t.start();
     return  true ;
}

   这个是addIfUnderCorePoolSize方法的具体实现,从名字能够看出它的意图就是当低于核心吃大小时执行的方法。下面看其具体实现,首先获取到锁,由于这地方涉及到线程池状态的变化,先经过if语句判断当前线程池中的线程数目是否小于核心池大小,有朋友也许会有疑问:前面在execute()方法中不是已经判断过了吗,只有线程池当前线程数目小于核心池大小才会执行addIfUnderCorePoolSize方法的,为什么这地方还要继续判断?缘由很简单,前面的判断过程当中并无加锁,所以可能在execute方法判断的时候poolSize小于corePoolSize,而判断完以后,在其余线程中又向线程池提交了任务,就可能致使poolSize不小于corePoolSize了,因此须要在这个地方继续判断。而后接着判断线程池的状态是否为RUNNING,缘由也很简单,由于有可能在其余线程中调用了shutdown或者shutdownNow方法。而后就是执行

1
t = addThread(firstTask);

   这个方法也很是关键,传进去的参数为提交的任务,返回值为Thread类型。而后接着在下面判断t是否为空,为空则代表建立线程失败(即poolSize>=corePoolSize或者runState不等于RUNNING),不然调用t.start()方法启动线程。

  咱们来看一下addThread方法的实现:

1
2
3
4
5
6
7
8
9
10
11
12
private  Thread addThread(Runnable firstTask) {
     Worker w =  new  Worker(firstTask);
     Thread t = threadFactory.newThread(w);   //建立一个线程,执行任务   
     if  (t !=  null ) {
         w.thread = t;             //将建立的线程的引用赋值为w的成员变量       
         workers.add(w);
         int  nt = ++poolSize;      //当前线程数加1       
         if  (nt > largestPoolSize)
             largestPoolSize = nt;
     }
     return  t;
}

   在addThread方法中,首先用提交的任务建立了一个Worker对象,而后调用线程工厂threadFactory建立了一个新的线程t,而后将线程t的引用赋值给了Worker对象的成员变量thread,接着经过workers.add(w)将Worker对象添加到工做集当中。

  下面咱们看一下Worker类的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
private  final  class  Worker  implements  Runnable {
     private  final  ReentrantLock runLock =  new  ReentrantLock();
     private  Runnable firstTask;
     volatile  long  completedTasks;
     Thread thread;
     Worker(Runnable firstTask) {
         this .firstTask = firstTask;
     }
     boolean  isActive() {
         return  runLock.isLocked();
     }
     void  interruptIfIdle() {
         final  ReentrantLock runLock =  this .runLock;
         if  (runLock.tryLock()) {
             try  {
         if  (thread != Thread.currentThread())
         thread.interrupt();
             finally  {
                 runLock.unlock();
             }
         }
     }
     void  interruptNow() {
         thread.interrupt();
     }
 
     private  void  runTask(Runnable task) {
         final  ReentrantLock runLock =  this .runLock;
         runLock.lock();
         try  {
             if  (runState < STOP &&
                 Thread.interrupted() &&
                 runState >= STOP)
             boolean  ran =  false ;
             beforeExecute(thread, task);    //beforeExecute方法是ThreadPoolExecutor类的一个方法,没有具体实现,用户能够根据
             //本身须要重载这个方法和后面的afterExecute方法来进行一些统计信息,好比某个任务的执行时间等           
             try  {
                 task.run();
                 ran =  true ;
                 afterExecute(task,  null );
                 ++completedTasks;
             catch  (RuntimeException ex) {
                 if  (!ran)
                     afterExecute(task, ex);
                 throw  ex;
             }
         finally  {
             runLock.unlock();
         }
     }
 
     public  void  run() {
         try  {
             Runnable task = firstTask;
             firstTask =  null ;
             while  (task !=  null  || (task = getTask()) !=  null ) {
                 runTask(task);
                 task =  null ;
             }
         finally  {
             workerDone( this );    //当任务队列中没有任务时,进行清理工做       
         }
     }
}

   它实际上实现了Runnable接口,所以上面的Thread t = threadFactory.newThread(w);效果跟下面这句的效果基本同样:

1
Thread t =  new  Thread(w);

   至关于传进去了一个Runnable任务,在线程t中执行这个Runnable。

  既然Worker实现了Runnable接口,那么天然最核心的方法即是run()方法了:

1
2
3
4
5
6
7
8
9
10
11
12
public  void  run() {
     try  {
         Runnable task = firstTask;
         firstTask =  null ;
         while  (task !=  null  || (task = getTask()) !=  null ) {
             runTask(task);
             task =  null ;
         }
     finally  {
         workerDone( this );
     }
}

   从run方法的实现能够看出,它首先执行的是经过构造器传进来的任务firstTask,在调用runTask()执行完firstTask以后,在while循环里面不断经过getTask()去取新的任务来执行,那么去哪里取呢?天然是从任务缓存队列里面去取,getTask是ThreadPoolExecutor类中的方法,并非Worker类中的方法,下面是getTask方法的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
Runnable getTask() {
     for  (;;) {
         try  {
             int  state = runState;
             if  (state > SHUTDOWN)
                 return  null ;
             Runnable r;
             if  (state == SHUTDOWN)   // Help drain queue
                 r = workQueue.poll();
             else  if  (poolSize > corePoolSize || allowCoreThreadTimeOut)  //若是线程数大于核心池大小或者容许为核心池线程设置空闲时间,
                 //则经过poll取任务,若等待必定的时间取不到任务,则返回null
                 r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
             else
                 r = workQueue.take();
             if  (r !=  null )
                 return  r;
             if  (workerCanExit()) {     //若是没取到任务,即r为null,则判断当前的worker是否能够退出
                 if  (runState >= SHUTDOWN)  // Wake up others
                     interruptIdleWorkers();    //中断处于空闲状态的worker
                 return  null ;
             }
             // Else retry
         catch  (InterruptedException ie) {
             // On interruption, re-check runState
         }
     }
}

   在getTask中,先判断当前线程池状态,若是runState大于SHUTDOWN(即为STOP或者TERMINATED),则直接返回null。

  若是runState为SHUTDOWN或者RUNNING,则从任务缓存队列取任务。

  若是当前线程池的线程数大于核心池大小corePoolSize或者容许为核心池中的线程设置空闲存活时间,则调用poll(time,timeUnit)来取任务,这个方法会等待必定的时间,若是取不到任务就返回null。

  而后判断取到的任务r是否为null,为null则经过调用workerCanExit()方法来判断当前worker是否能够退出,咱们看一下workerCanExit()的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private  boolean  workerCanExit() {
     final  ReentrantLock mainLock =  this .mainLock;
     mainLock.lock();
     boolean  canExit;
     //若是runState大于等于STOP,或者任务缓存队列为空了
     //或者  容许为核心池线程设置空闲存活时间而且线程池中的线程数目大于1
     try  {
         canExit = runState >= STOP ||
             workQueue.isEmpty() ||
             (allowCoreThreadTimeOut &&
              poolSize > Math.max( 1 , corePoolSize));
     finally  {
         mainLock.unlock();
     }
     return  canExit;
}

   也就是说若是线程池处于STOP状态、或者任务队列已为空或者容许为核心池线程设置空闲存活时间而且线程数大于1时,容许worker退出。若是容许worker退出,则调用interruptIdleWorkers()中断处于空闲状态的worker,咱们看一下interruptIdleWorkers()的实现:

1
2
3
4
5
6
7
8
9
10
void  interruptIdleWorkers() {
     final  ReentrantLock mainLock =  this .mainLock;
     mainLock.lock();
     try  {
         for  (Worker w : workers)   //实际上调用的是worker的interruptIfIdle()方法
             w.interruptIfIdle();
     finally  {
         mainLock.unlock();
     }
}

   从实现能够看出,它实际上调用的是worker的interruptIfIdle()方法,在worker的interruptIfIdle()方法中:

1
2
3
4
5
6
7
8
9
10
11
12
void  interruptIfIdle() {
     final  ReentrantLock runLock =  this .runLock;
     if  (runLock.tryLock()) {     //注意这里,是调用tryLock()来获取锁的,由于若是当前worker正在执行任务,锁已经被获取了,是没法获取到锁的
                                 //若是成功获取了锁,说明当前worker处于空闲状态
         try  {
     if  (thread != Thread.currentThread())  
     thread.interrupt();
         finally  {
             runLock.unlock();
         }
     }
}

    这里有一个很是巧妙的设计方式,假如咱们来设