您的位置: 首页  >  文章  >  【转】线程池ThreadPoolExecutor(四)

【转】线程池ThreadPoolExecutor(四)

分类: 文章 • 2024-11-23 21:00:58

【转】线程池ThreadPoolExecutor(四)

1、构造方法

请参考上篇文章：http://blog.****.net/ochangwen/article/details/53044733

2、源码详解

线程池内部有一些状态，先来了解下这些状态的机制。以下用代码注释的方式来解释其中的含义。
[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. /* 
  2. 这个是用一个int来表示workerCount和runState的，其中runState占int的高3位， 
  3. 其它29位为workerCount的值。 
  4.  
  5. workerCount:当前活动的线程数； 
  6. runState：线程池的当前状态。 
  7.  
  8. 用AtomicInteger是因为其在并发下使用compareAndSet效率非常高； 
  9. 当改变当前活动的线程数时只对低29位操作，如每次加一减一，workerCount的值变了， 
  10. 但不会影响高3位的runState的值。当改变当前状态的时候，只对高3位操作，不会改变低29位的计数值。 
  11. 这里有一个假设，就是当前活动的线程数不会超过29位能表示的值，即不会超过536870911， 
  12. 就目前以及可预见的很长一段时间来讲，这个值是足够用了 
  13. */  
  14. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));  
  15.   
  16. //COUNT_BITS，就是用来表示workerCount占用一个int的位数，其值为前面说的29  
  17. private static final int COUNT_BITS = Integer.SIZE - 3;  
  18.   
  19. /* 
  20. CAPACITY为29位能表示的最大容量，即workerCount实际能用的最大值。 
  21. 其值的二进制为：00011111111111111111111111111111（占29位，29个1） 
  22. */  
  23. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;  
  24.   
  25. /* 
  26. 以下常量是线程池的状态，状态存储在int的高3位，所以要左移29位。 
  27. 腾出的低29位来表示workerCount 
  28. 注意，这5个状态是有大小关系的。RUNNING<shutdown<stop<tidying<terminated 当需要判断多个状态时，只需要用<或="">来判断就可以了 
  29. */  
  30.   
  31. /* 
  32. RUNNING的含义：线程池能接受新任务，并且可以运行队列中的任务 
  33. -1的二进制为32个1，移位后为：11100000000000000000000000000000 
  34. */  
  35. private static final int RUNNING    = -1 << COUNT_BITS;  
  36.   
  37. /* 
  38. SHUTDOWN的含义：不再接受新任务，但仍可以执行队列中的任务 
  39. 0的二进制为32个0，移位后还是全0 
  40. */  
  41. private static final int SHUTDOWN   =  0 << COUNT_BITS;  
  42.   
  43. /* 
  44. STOP的含义：不再接受新任务，不再执行队列中的任务，而且要中断正在处理的任务 
  45. 1的二进制为前面31个0，最后一个1，移位后为：00100000000000000000000000000000 
  46. */  
  47. private static final int STOP       =  1 << COUNT_BITS;  
  48.   
  49. /* 
  50. TIDYING的含义：所有任务均已终止，workerCount的值为0， 
  51. 转到TIDYING状态的线程即将要执行terminated()钩子方法. 
  52. 2的二进制为00000000000000000000000000000010 
  53. 移位后01000000000000000000000000000000 
  54. */  
  55. private static final int TIDYING    =  2 << COUNT_BITS;  
  56.   
  57. /* 
  58. TERMINATED的含义:terminated()方法执行结束. 
  59. 3的二进制为00000000000000000000000000000011 
  60. 移位后01100000000000000000000000000000 
  61. */  
  62. private static final int TERMINATED =  3 << COUNT_BITS;  
[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. 各状态之间可能的转变有以下几种：  
  2. RUNNING -> SHUTDOWN  
  3.     调用了shutdown方法，线程池实现了finalize方法，在里面调用了shutdown方法，因此shutdown可能是在finalize中被隐式调用的  
  4. (RUNNING or SHUTDOWN) -> STOP  
  5.     调用了shutdownNow方法  
  6. SHUTDOWN -> TIDYING  
  7.     当队列和线程池均为空的时候  
  8. STOP -> TIDYING  
  9.     当线程池为空的时候  
  10. TIDYING -> TERMINATED  
  11.     terminated()钩子方法调用完毕  
[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. /* 
  2. 传入的参数为存储runState和workerCount的int值，这个方法用于取出runState的值。 
  3. ~为按位取反操作，~CAPACITY值为：11100000000000000000000000000000， 
  4. 再同参数做&操作，就将低29位置0了，而高3位还是保持原先的值，也就是runState的值 
  5. */  
  6. private static int runStateOf(int c)     { return c & ~CAPACITY; }  
  7.   
  8. /* 
  9. 传入的参数为存储runState和workerCount的int值，这个方法用于取出workerCount的值。 
  10. 因为CAPACITY值为：00011111111111111111111111111111，所以&操作将参数的高3位置0了， 
  11. 保留参数的低29位，也就是workerCount的值。 
  12. */  
  13. private static int workerCountOf(int c)  { return c & CAPACITY; }  
  14.   
  15. /* 
  16. 将runState和workerCount存到同一个int中，这里的rs就是runState， 
  17. 是已经移位过的值，填充返回值的高3位，wc填充返回值的低29位 
  18. */  
  19. private static int ctlOf(int rs, int wc) { return rs | wc; }  


接下来通过状态来看线程池的运行：
完成线程的创建后，
首先通过 submit 或者 execute 方法提交，实际submit 方法也是内部调用 execute方法；
submit方法的实现是在AbstractExecutorService类里，三种重载方法类似，

[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. public Future<?> submit(Runnable task) {  
  2.     if (task == null) throw new NullPointerException();  
  3.     RunnableFuture<Void> ftask = newTaskFor(task, null);  
  4.     execute(ftask); //<span style="font-size:12px;">内部调用 execute方法</span>  
  5.     return ftask;  
  6. }  
execute方法是在ThreadPoolExecutor类里实现
[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. public void execute(Runnable command) {  
  2.     if (command == null)  
  3.         throw new NullPointerException();  
  4.   
  5.     int c = ctl.get();  
  6.     if (workerCountOf(c) < corePoolSize) {  
  7.         if (addWorker(command, true))  
  8.             return;  
  9.         c = ctl.get();  
  10.     }  
  11.     if (isRunning(c) && workQueue.offer(command)) {  
  12.         int recheck = ctl.get();  
  13.         if (! isRunning(recheck) && remove(command))  
  14.             reject(command);  
  15.         else if (workerCountOf(recheck) == 0)  
  16.             addWorker(null, false);  
  17.     }  
  18.     else if (!addWorker(command, false))  
  19.         reject(command);  
  20. }  
    1、判断当前的运行的线程数是否小于corePoolSize如果是，使用入参任务通过addWord方法创建一个新的线程，如果能完成新线程创建exexute方法结束，成功提交任务；
    2、在第一步没有完成任务提交；状态为运行并且能否成功加入任务到工作队列后，再进行一次check，如果状态在任务加入队列后变为了非运行（有可能是在执行到这里线程池shutdown了），非运行状态下当然是需要reject；然后再判断当前线程数是否为0（有可能这个时候线程数变为了0），如是，新增一个线程；
    3、如果不能加入任务到工作队列，将尝试使用任务新增一个线程，如果失败，则是线程池已经shutdown或者线程池已经达到饱和状态，所以reject;
从上面新增任务的execute方法也可以看出，拒绝策略不仅仅是在饱和状态下使用，在线程池进入到关闭阶段同样需要使用到；

  1.先判断传入的参数command是否为null，为null就抛出NullPointerException。
  2.然后通过workerCountOf方法从ctl所表示的int值中提取出低29位的值，也就是当前活动的线程数。
    private static int workerCountOf(int c)  { return c & CAPACITY; }
  3.如果当前活动的线程数小于corePoolSize，则增加一个线程（addWorker，接下来会讲解这个方法，其返回值表示是否新增线程成功）来执行新传入的任务。什么概念？也就是说当池中的线程数小于corePoolSize的时候，不管池中的线程是否有空闲的，每次调用该方法都去增加一个线程，直到池中的数目达到corePoolSize为止。
  4.如果新增线程成功，则由新线程执行传入的任务command。在这里有可能出现增加线程失败的情况（原因在解释addWorker的时候讲），那就要当做池中当前线程数超过corePoolSize的情况进行处理。
  5.也就是进入第三个if里，若当前线程池的状态为RUNNING状态，且将任务command加入队列成功，就会执行if内的逻辑。


先讲此if对应的else里的情况，若在执行execute的时候同时有其它线程执行了shutdown方法，而这两个方法不是互斥的，就有竞态条件问题，execute方法之前判断状态为RUNNING，而执行了几条语句后可能池的状态已经变掉了，因此，如果池的状态不为RUNNING或在将command加入队列失败的时候（失败可能是有界队列满了），两种情况要分开处理，

当只是状态仍为RUNNING，而队列满的时候，若池中当前活动的线程数小于maximumPoolSize，则会往池中添加线程，若添线程数已经达到了maximumPoolSize或其它原因导致新增线程失败，就会拒绝该任务（reject(command)）。

当状态不为RUNNING的时候，if里的addWorker(command, false)操作将直接返回false，使得if条件为true，也会拒绝这个任务。再继续前面往队列里加入任务成功的处理方式。加入任务成功后，会再次检测池的状态是否为RUNNING，若不是，则从池中移出并拒绝该任务，这也就是说，当池的被SHUTDOWN后，将不再接受新任务。这些检查若都没问题，还需要看看池中的活动线程数有没有变成0（执行的任务抛出异常等导致），若为0，则往里加入一个线程，该线程回去队列里拿任务执行。如果一次性往队列里提交了很多任务，而池中的每个任务执行都抛出异常，那么会不会导致剩余的任务得不到执行？显然不会，这个在后面再讲。

接下来介绍上面提到的addWorker方法，方法实现如下：

[java] view plain copy
 
 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)
  1. private static int runStateOf(int c)     { return c & ~CAPACITY; }  
  2.   
  3. //使用原子的compareAndSet来替换旧值。但并不保证成功，若成功，该方法返回true；若失败，则返回false  
  4. private boolean compareAndIncrementWorkerCount(int expect) {  
  5.     return ctl.compareAndSet(expect, expect + 1);  
  6. }  
  7.   
  8. private final ReentrantLock mainLock = new ReentrantLock();  
  9.   
  10. /** 
  11.  * @param firstTask 用于指定新增的线程执行的第一个任务 
  12.  * @param core true表示在新增线程时会判断当前活动线程数是否少于corePoolSize， 
  13.  *             false表示新增线程前需要判断当前活动线程数是否少于maximumPoolSize。 
  14.  * @return  该方法的返回值代表是否成功新增一个线程 
  15.  */  
  16. private boolean addWorker(Runnable firstTask, boolean core) {  
  17.     retry:  // 标签，写代码时避免使用标签。  
  18.     for (;;) {  
  19.         int c = ctl.get();  
  20.         int rs = runStateOf(c); // 取出存储在ctl中的状态值  
  21.   
  22.         // Check if queue empty only if necessary.  
  23.         /* 
  24.         转换成一个等价实现： 
  25.         rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null || workQueue.isEmpty())， 
  26.  
  27.         rs != SHUTDOWN结合前面的rs >= SHUTDOWN，表示线程池的状态已经由SHUTDOWN转为剩余的三个状态之一了，此时就要拒绝这个传入的任务； 
  28.         括号里的第二个条件表示状态已经为非运行状态了，却传入了一个任务，这个任务也要拒绝； 
  29.         括号里的第三个条件表示线程池的状态不为RUNNING，但队列中没有任务了，就不需要新增线程了。 
  30.          */  
  31.         if (rs >= SHUTDOWN && //如果已经调用了SHUTDOWN，池的状态改变后，这里就为true  
  32.                 ! (rs == SHUTDOWN &&  
  33.                         firstTask == null &&  
  34.                         ! workQueue.isEmpty()))  
  35.             return false;  
  36.   
  37.         for (;;) {  
  38.             int wc = workerCountOf(c); // 取出当前活动的线程数。  
  39.             /* 
  40.             若当前活动线程数超过低29位能表示的最大值（也就是容量）时就不能再加线程了，因为再加就会影响状态的值了！ 
  41.  
  42.             若传入的参数core参数为true，则当前活动的线程数要小于corePoolSize才能创建新线程，大于或等于corePoolSize就不能再创建了； 
  43.             若core参数为false，则当前活动的线程数要小于maximumPoolSize才能创建新线程，大于或等于maximumPoolSize就不能再创建了。 
  44.              */  
  45.             if (wc >= CAPACITY ||  
  46.                     wc >= (core ? corePoolSize : maximumPoolSize))  
  47.                 return false;  
  48.             if (compareAndIncrementWorkerCount(c)) // 使用CAS操作将当前活动线程数加一  
  49.                 break retry; // 当加一成功，则跳出大循环，进入循环体后面的真正新增线程的地方；  
  50.             c = ctl.get();  // Re-read ctl  
  51.   
  52.             //若加一不成功，判断下当前状态改变没有，若改变了则重新开始外层循环的下一次迭代，  
  53.             // 若状态没有改变，只是加一失败，那么就继续内层循环，直到加一成功。  
  54.             /* 
  55.             往当前活动线程数加一成功后，就会来真的新增线程了 
  56.             （先加一后新增线程可以避免锁的使用，使用CAS原子操作加一后，其它线程看到的就是加一后的值， 
  57.             若达到上限，其它线程就不会去创建新线程了。若先创建线程，再去加一，若不加锁， 
  58.             假如一个使用无界队列的线程池，当前活动线程数为corePoolSize少一， 
  59.             外部线程在执行execute的时候都发现线程数不足corePoolSize，都去创建线程， 
  60.             而最终只能有一个线程进入线程池，其它的都得作废，而加锁可以解决这个问题，但是降低了线程池的可伸缩性）。 
  61.              */  
  62.             if (runStateOf(c) != rs)  
  63.                 continue retry;  
  64.             // else CAS failed due to workerCount change; retry inner loop  
  65.         }  
  66.     }  
  67.   
  68.     boolean workerStarted = false;  
  69.     boolean workerAdded = false;  
  70.     ThreadPoolExecutor.Worker w = null;  
  71.     try {  
  72.         final ReentrantLock mainLock = this.mainLock;  
  73.         w = new ThreadPoolExecutor.Worker(firstTask); //  Worker为内部类，封装了线程和任务，通过ThreadFactory创建线程，可能失败抛异常或者返回null  
  74.         final Thread t = w.thread;  
  75.         if (t != null) {  
  76.             mainLock.lock();  
  77.             try {  
  78.                 // Recheck while holding lock.  
  79.                 // Back out on ThreadFactory failure or if  
  80.                 // shut down before lock acquired.  
  81.                 int c = ctl.get();  
  82.                 int rs = runStateOf(c);  
  83.   
  84.                 if (rs < SHUTDOWN ||  
  85.                         (rs == SHUTDOWN && firstTask == null)) {  
  86.                     if (t.isAlive()) // precheck that t is startable  
  87.                         throw new IllegalThreadStateException();  
  88.                     workers.add(w);  
  89.                     int s = workers.size();  
  90.                     if (s > largestPoolSize)  
  91.                         largestPoolSize = s;  
  92.                     workerAdded = true;  
  93.                 }  
  94.             } finally {  
  95.                 mainLock.unlock();  
  96.             }  
  97.             if (workerAdded) {  
  98.                 t.start();  
  99.                 workerStarted = true;  
  100.             }  
  101.         }  
  102.     } finally {  
  103.         if (! workerStarted)  
  104.             addWorkerFailed(w);  
  105.     }  
  106.     return workerStarted;  
  107. }  


    真正新增线程的地方分析。接下来看如何新增线程的。Worker w = new Worker(firstTask)，在Worker的构造方法中，创建了一个线程对象，但这个线程是没有启动的。在构造方法中启动线程，会导致this对象泄露，让线程看到未完整构建的对象，这个要避免。既然不能在构造方法里启动，那么就把创建的线程对象拿出来吧，也就是赋给了t变量。因为整个过程并不是互斥的，所以创建完线程对象后再来判断下当前池的状态，接下来的判断又比较复杂：t == null || (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null))，转换成一个容易看懂的等价实现：t == null || (rs >= SHUTDOWN && (rs != SHUTDOWN || firstTask != null))。里面有个t==null？为啥会出现t==null? Worker的构造方法是通过调用getThreadFactory().newThread(this)方法来创建线程的，而newThread方法可能会返回null（threadFactory可以通过ThreadPoolExecutor的构造方法传入，如没有传入，有个默认实现）。当创建线程失败要减少当前活动线程数；当池的状态非RUNNING和SHUTDOWN的时候，也需要减少当前活动线程数，并要尝试终止线程池；当线程池的状态为非RUNNING，且有初始任务要执行的时候，因为这个状态要拒绝新进来的任务，所以这个新增的线程也没有用处了。当状态判断没有问题时，就会将创建的Worker对象加入到workers字段中（线程终止时会从里面移除，后面会讲到），当池中的工作线程创新高时，会将这个数记录到largestPoolSize字段中。然后就可以启动这个线程t了。若start后，状态又变成了SHUTDOWN状态（如调用了shutdownNow方法）且新建的线程没有被中断过，就要中断该线程（shutdownNow方法要求中断正在执行的线程），shutdownNow方法本身也会去中断存储在workers中的所有线程，为什么这里还有自己处理下呢？中断所有线程的时候需要持有mainLock锁，而添加Worker对象到workers字段中也要持有mainLock锁，所以存在这样一种很难出现的场景：在将Worker对象加入workers字段，释放mainLock锁之后，Worker对象中的线程（即t）启动前，shutdownNow获得了mainLock锁并完成了所有中断操作，而当线程对象还没调用start之前调用该线程的interrupt方法是无效的。所以，t启动后的这段小代码就是为了防止这种极端情况的出现。


在继续其他方法之前，先说下Worker这个内部类。我们看一下每次新增一个线程后这个线程都做了些什么，显然需要看看Worker的run方法：







[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. /** 
    2. 

  2.  * Worker实现了Runnable接口，可以在后续作为Thread的构造方法参数用以创建线程。 
    3. 

  3.  * 同时,Worker还继承了AbstractQueuedSynchronizer类，只是简化每个Worker对象相关的锁的获取，在每次执行一个任务的时候，都需要持有这个锁。 
    4. 

  4.  * 
    5. 

  5.  * 在以前的ThreadPoolExecutor实现中，并没有继承AbstractQueuedSynchronizer， 
    6. 

  6.  * 而是在Worker内部声明了一个对象字段private final ReentrantLock runLock = new ReentrantLock()， 
    7. 

  7.  * 每次执行一个任务的时候，需要对runLock加锁。 
    8. 

  8.  */  
    9. 

  9. private final class Worker  
    10. 

  10.         extends AbstractQueuedSynchronizer  
    11. 

  11.         implements Runnable {  
    12. 

  12.     ......  
    13. 

  13.     public void run() {  
    14. 

  14.         runWorker(this);  
    15. 

  15.     }  
    16. 

  16.     ......  
    17. 

  17. }  
    18. 




只是简单的调用了runWorker方法，继续看runWorker：


    这个方法逻辑很简单。还记得前面提到的新增线程时指定第一个任务吗？若存在第一个任务，则先执行第一个任务，否则，从队列中拿任务，不断的执行，直到getTask返回null或执行任务出错（中断或任务本身抛出异常），就退出while循环。getTask方法后面会详细讲解。当有任务执行时（之前通过参数传入的第一个任务或从队列中获取的任务），需要做一个状态判断。也就是clearInterruptsForTaskRun方法，来看看这个方法干了什么：(这个原来的方法就是下面if里的条件)





[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. final void runWorker(ThreadPoolExecutor.Worker w) {  
    2. 

  2.     Thread wt = Thread.currentThread();  
    3. 

  3.     Runnable task = w.firstTask;  
    4. 

  4.     w.firstTask = null;  
    5. 

  5.     w.unlock(); // allow interrupts  
    6. 

  6.     boolean completedAbruptly = true;  
    7. 

  7.     try {  
    8. 

  8.         while (task != null || (task = getTask()) != null) {  
    9. 

  9.             w.lock();  
    10. 

  10.   
    11. 

  11.             /* 
    12. 

  12.             这里利用了&&的短路特性，当前一个条件为true的时候才去执行后面一个条件。 
    13. 

  13.  
    14. 

  14.             当当前状态小于STOP时，也就是当前状态为RUNNING时，需要清除线程的中断状态 
    15. 

  15.             （线程池为RUNNING状态，线程却的中断状态却为true， 
    16. 

  16.             可能在上次执行的任务里调用了类似Thread.currentThread().interrupt()的方法， 
    17. 

  17.             因此当然不能让接下来执行的任务受之前任务的影响）， 
    18. 

  18.  
    19. 

  19.             如果Thread.interrupted()返回false，表示以前没有设置过中断，整个if的结果就是false； 
    20. 

  20.  
    21. 

  21.             如果Thread.interrupted()返回true，那就要考虑为什么会是true了。 
    22. 

  22.             是RUNNING状态就已经被中断了还是判断第一个条件后另外一个非池中的线程调用了shutdownNow中断了该线程？如 
    23. 

  23.             果是后者，表示正在执行的任务需要中断，所以第三个条件判断当前池的状态是否不为RUNNING， 
    24. 

  24.  
    25. 

  25.             如果不为RUNNING，那么就要重新中断该线程以维护shutdownNow方法的语义。 
    26. 

  26.              */  
    27. 

  27.             if ((runStateAtLeast(ctl.get(), STOP) ||  
    28. 

  28.                     (Thread.interrupted() &&  
    29. 

  29.                             runStateAtLeast(ctl.get(), STOP))) &&  
    30. 

  30.                     !wt.isInterrupted())  //这个if条件就是原来的clearInterruptsForTaskRun  
    31. 

  31.                 wt.interrupt();  
    32. 

  32.             try {  
    33. 

  33.                 //在真正执行任务前，调用该方法。这是一个钩子方法，用户可以继承ThreadPoolExecutor重写beforeExecute方法来做一些事情。  
    34. 

  34.                 beforeExecute(wt, task);  
    35. 

  35.                 Throwable thrown = null;  
    36. 

  36.                 try {  
    37. 

  37.                     task.run();  
    38. 

  38.                 } catch (RuntimeException x) {  
    39. 

  39.                     thrown = x; throw x;  
    40. 

  40.                 } catch (Error x) {  
    41. 

  41.                     thrown = x; throw x;  
    42. 

  42.                 } catch (Throwable x) {  
    43. 

  43.                     thrown = x; throw new Error(x);  
    44. 

  44.                 } finally {  
    45. 

  45.                     afterExecute(task, thrown); //也是个钩子方法，同beforeExecute方法。  
    46. 

  46.                 }  
    47. 

  47.             } finally {  
    48. 

  48.                 /* 
    49. 

  49.                 随后将task变量置为null，让其再从队列里接收任务， 
    50. 

  50.                 若不置为null，就满足while的第一个条件了，结果就是这个任务被死循环执行 
    51. 

  51.                  */  
    52. 

  52.                 task = null;  
    53. 

  53.                 w.completedTasks++; // 然后将该线程完成的任务数自增  
    54. 

  54.                 w.unlock();  
    55. 

  55.                 /* 
    56. 

  56.                 只有当线程终止的时候，才会将该线程执行的任务总数加到线程池的completedTaskCount中， 
    57. 

  57.                 所以completedTaskCount这个值并不是一个准确值。 
    58. 

  58.                  */  
    59. 

  59.             }  
    60. 

  60.         }  
    61. 

  61.         /* 
    62. 

  62.         在最后有一个将completedAbruptly置为false的操作， 
    63. 

  63.         如果线程能走到这里来，说明该线程在执行任务过程中没有抛出异常，也就是说该线程并不是异常结束的，而是正常结束的； 
    64. 

  64.         如果走不到这一步，completedAbruptly的值还是初始值true，表示线程是异常结束的。 
    65. 

  65.          */  
    66. 

  66.         completedAbruptly = false;  
    67. 

  67.     } finally {  
    68. 

  68.         // 线程结束时，会调用processWorkerExit方法做一些清理和数据同步的工作：  
    69. 

  69.         processWorkerExit(w, completedAbruptly);  
    70. 

  70.     }  
    71. 

  71. }  
    72. 

  72.   
    73. 

  73. private static boolean runStateAtLeast(int c, int s) {  
    74. 

  74.     return c >= s;  
    75. 

  75. }  
    76. 

  76.   
    77. 

  77. protected void beforeExecute(Thread t, Runnable r) { }  
    78. 

  78.   
    79. 

  79. protected void afterExecute(Runnable r, Throwable t) { }  
    80. 











线程结束时，会调用processWorkerExit方法做一些清理和数据同步的工作：







[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. private void processWorkerExit(ThreadPoolExecutor.Worker w, boolean completedAbruptly) {  
    2. 

  2.     if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted  
    3. 

  3.         decrementWorkerCount();  
    4. 

  4.   
    5. 

  5.     final ReentrantLock mainLock = this.mainLock;  
    6. 

  6.     mainLock.lock();  
    7. 

  7.     try {  
    8. 

  8.         completedTaskCount += w.completedTasks;  
    9. 

  9.         workers.remove(w);  
    10. 

  10.     } finally {  
    11. 

  11.         mainLock.unlock();  
    12. 

  12.     }  
    13. 

  13.   
    14. 

  14.     tryTerminate();  
    15. 

  15.   
    16. 

  16.     int c = ctl.get();  
    17. 

  17.     if (runStateLessThan(c, STOP)) {  
    18. 

  18.         if (!completedAbruptly) {  
    19. 

  19.             int min = allowCoreThreadTimeOut ? 0 : corePoolSize;  
    20. 

  20.             if (min == 0 && ! workQueue.isEmpty())  
    21. 

  21.                 min = 1;  
    22. 

  22.             if (workerCountOf(c) >= min)  
    23. 

  23.                 return; // replacement not needed  
    24. 

  24.         }  
    25. 

  25.         addWorker(null, false);  
    26. 

  26.     }  
    27. 

  27. }  
    28. 




    如果线程是异常结束（被中断、任务执行本身异常等），当前活动的线程数减少一个。如果是正常结束的呢？不应该将其也减一吗？不用担心，在runWorker的while最后一次循环中的getTask方法里做掉了。
    接下来将该线程执行过的任务数加到completedTaskCount中，这个在前面也提到了。然后从workers中去除该工作线程。如果该线程的中断是因为调用了shutdown、shutdownNow接口而中断的该如何处理？就是这个tryTerminate了，来看下tryTerminate干了什么：



[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. final void tryTerminate() {  
    2. 

  2.     for (;;) {  
    3. 

  3.         int c = ctl.get();  
    4. 

  4.         if (isRunning(c) ||  
    5. 

  5.                 runStateAtLeast(c, TIDYING) ||  
    6. 

  6.                 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))  
    7. 

  7.             return;  
    8. 

  8.         if (workerCountOf(c) != 0) { // Eligible to terminate  
    9. 

  9.             interruptIdleWorkers(ONLY_ONE);  
    10. 

  10.             return;  
    11. 

  11.         }  
    12. 

  12.   
    13. 

  13.         final ReentrantLock mainLock = this.mainLock;  
    14. 

  14.         mainLock.lock();  
    15. 

  15.         try {  
    16. 

  16.             if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {  
    17. 

  17.                 try {  
    18. 

  18.                     terminated();  
    19. 

  19.                 } finally {  
    20. 

  20.                     ctl.set(ctlOf(TERMINATED, 0));  
    21. 

  21.                     termination.signalAll();  
    22. 

  22.                 }  
    23. 

  23.                 return;  
    24. 

  24.             }  
    25. 

  25.         } finally {  
    26. 

  26.             mainLock.unlock();  
    27. 

  27.         }  
    28. 

  28.         // else retry on failed CAS  
    29. 

  29.     }  
    30. 

  30. }  
    31. 







这个方法用来干啥的？当池的状态为SHUTDOWN且任务队列为空，需要将池的状态转变为TERMINATED；当池的状态为STOP且池中的当前活动线程数为0，要将池的状态转换成TERMINATED。这个方法就是用来做这种状态转变的。如果状态是RUNNING，表示线程池还正在提供服务，不需要状态变换；如果状态为TIDYING或TERMINATED，池中的活动线程数已经是0，自然也不需要做什么操作了；若状态为SHUTDWON，但队列中还有任务，此时这些任务还需要做掉，因此池中的线程不能终止，因此，这种情况下也不需要做什么。如果状态为SHUTDWON但队列中已经没有任务了，这里调用了一个interruptIdleWorkers(ONLY_ONE)操作去中断一个空闲线程。这么做是为什么？【关于这个的理解可能有问题】调用这个方法的目的是将shutdown信号传播给其它线程。调用shutdown方法的时候会去中断所有空闲线程，如果这时候池中所有的线程都正在执行任务，那么就不会有线程被中断，调用shutdown方法只是设置了线程池的状态为SHUTDOWN，在取任务(getTask,后面会细说)的时候，假如很多线程都发现队列里还有任务（没有使用锁，存在竞态条件），然后都去调用take，如果任务数小于池中的线程数，那么必然有方法调用take后会一直等待（shutdown的时候这些线程正在执行任务，所以没能调用它的interrupt，其中断状态没有被设置），那么在没有任务且线程池的状态为SHUTDWON的时候，这些等待中的空闲线程就需要被终止iinterruptIdleWorkers(ONLY_ONE)回去中断一个线程，让其从take中退出，然后这个线程也进入同样的逻辑，去终止一个其它空闲线程，直到池中的活动线程数为0。


当状态为SHUTDOWN，且活动线程数为0的时候，就可以进入TIDYING状态了，进入TIDYING状态就可以执行钩子方法terminated()，该方法执行结束就进入了TERMINATED状态（参考前文中各状态的含义以及可能的状态转变）。最后的termination.signalAll()所为何事？当线程池shutdown后，外部可能还有很多线程在等待线程池真正结束，即调用了awaitTermination方法，该方法中，外部线程就是在termination上await的，所以，线程池关闭之前要唤醒这些等待的线程，告诉它们线程池关闭结束了。


继续说processWorkerExit方法中调用tryTerminate之后的代码。如果池的状态仍为RUNNING，而线程是因为执行的任务本身抛出了异常而结束或正常结束时该如何处理？这时候池的状态还是RUNNING呢！那就是接下来的这个if块要做的事儿了。当池的状态还是RUNNING，又要分两种情况，一种是异常结束，一种是正常结束。异常结束比较好弄，直接加个线程替换死掉的线程就好了，也就是最后的addWorker操作。而正常结束又有几种情况了，如果允许core线程超时，也就是allowCoreThreadTimeOut为true，那么在池中没有任务的时候，调用带有时限参数的poll方法时就可能返回null，致使线程正常退出，如果允许core线程超时，池中最小的线程数可为0，如果此时队列又有任务了，那么池中必须要有一个线程，若池中活动的线程数不为0，就不需要新增线程来替换死掉的线程，否则就要新增一个；如果不允许core线程超时，池中的线程必须达到corePoolSize个才能让多的线程退出，而不需要用新的线程替换，否则也需要新增一个线程替换这个死掉的线程。


在runWorker执行任务之前调用了w.lock操作，为什么要在执行任务的时候锁定这个每个线程都有一份的锁呢？原因在于调用了线程池shutdown后（前面说过，SHUTDOWN的含义：不再接受新任务，但仍可以执行队列中的任务），会调用interruptIdleWorkers方法去终止空闲线程，该方法会持有mainLock锁，但此时队列中可能还有很多任务，线程也可能还正在执行任务，就可能有一些线程终止不掉。此时，有些线程可能刚执行任务结束，正准备再去队列中拿任务，有些可能还正在执行任务，有些可能刚拿到一个新的任务，对于仍进入队列中拿任务的线程，最终队列中任务会被拿完，而此时拿任务的线程会发现线程池的状态为SHUTDOWN，就会立马返回一个null，返回null意味着ThreadPoolExecutor.runWorker中的循环退出了，这个线程也就自动终止了；此外拿任务并没有持有mainLock锁，所以在终止空闲线程与线程非执行任务期间（如从队列获取任务）存在竞态条件。有可能已经判断了线程池的状态仍未RUNNING，准备从queue里take任务，而在执行take之前，另一个非池中的线程可能调用了shutdown，并且执行完了interruptIdleWorkers方法（马上就会介绍这个方法），若此时队列中恰好没有任务了，若这个正要调用take的线程阻塞，就不会醒过来了，不用担心，interruptIdleWorkers已经中断了该线程，而take是可以响应中断的，再调用take后会立马抛出异常。 对于正在执行中的任务，其它线程不能直接将这个正在线程中断掉，因此除了mainLock锁，interruptIdleWorkers还需要持有线程执行任务时获取的那把锁（这也是为什么执行任务的时候需要获取那个每个线程都有的锁的原因），如果获取不成功表示线程正在执行任务。看下终止空闲线程的方法实现：





[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. private void interruptIdleWorkers() {  
    2. 

  2.     interruptIdleWorkers(false);  
    3. 

  3. }  
    4. 

  4.   
    5. 

  5. private void interruptIdleWorkers(boolean onlyOne) {  
    6. 

  6.     final ReentrantLock mainLock = this.mainLock;  
    7. 

  7.     mainLock.lock();  
    8. 

  8.     try {  
    9. 

  9.         for (ThreadPoolExecutor.Worker w : workers) {  
    10. 

  10.             Thread t = w.thread;  
    11. 

  11.             if (!t.isInterrupted() && w.tryLock()) {  
    12. 

  12.                 try {  
    13. 

  13.                     t.interrupt();  
    14. 

  14.                 } catch (SecurityException ignore) {  
    15. 

  15.                 } finally {  
    16. 

  16.                     w.unlock();  
    17. 

  17.                 }  
    18. 

  18.             }  
    19. 

  19.             if (onlyOne)  
    20. 

  20.                 break;  
    21. 

  21.         }  
    22. 

  22.     } finally {  
    23. 

  23.         mainLock.unlock();  
    24. 

  24.     }  
    25. 

  25. }  
    26. 

  26. private void interruptWorkers() {  
    27. 

  27.     final ReentrantLock mainLock = this.mainLock;  
    28. 

  28.     mainLock.lock();  
    29. 

  29.     try {  
    30. 

  30.         for (ThreadPoolExecutor.Worker w : workers)  
    31. 

  31.             w.interruptIfStarted();  
    32. 

  32.     } finally {  
    33. 

  33.         mainLock.unlock();  
    34. 

  34.     }  
    35. 

  35. }  
    36. 





interruptIdleWorkers遍历workers中所有的工作线程，若线程没有被中断且能立即（tryLock）获取到前面提到的那把线程任务锁时，就中断该线程。为什么需要持有mainLock？mainLock是用来保护workers变量的。


shutdown是持有mainLock的，但是runWorker的时候并没有，那么，会不会出现碰巧出现同一时刻池中所有线程都刚好执行完任务，去取任务的时候发现池的状态为SHUTDOWN，就立即返回null并终止线程，而导致队列中的剩下的任务得不到执行？这是不会出现的，来看下getTask的实现：





[java] view plain copy


 

 【转】线程池ThreadPoolExecutor(四)【转】线程池ThreadPoolExecutor(四)



    

  1. private Runnable getTask() {  
    2. 

  2.     boolean timedOut = false; // Did the last poll() time out?  
    3. 

  3.   
    4. 

  4.     retry:  
    5. 

  5.     for (;;) {  
    6. 

  6.         int c = ctl.get();  
    7. 

  7.         int rs = runStateOf(c);  
    8. 

  8.   
    9. 

  9.         // Check if queue empty only if necessary.  
    10. 

  10.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {  
    11. 

  11.             decrementWorkerCount();  
    12. 

  12.             return null;  
    13. 

  13.         }  
    14. 

  14.   
    15. 

  15.         boolean timed;      // Are workers subject to culling?  
    16. 

  16.   
    17. 

  17.         for (;;) {  
    18. 

  18.             int wc = workerCountOf(c);  
    19. 

  19.             timed = allowCoreThreadTimeOut || wc > corePoolSize;  
    20. 

  20.   
    21. 

  21.             if (wc <= maximumPoolSize && ! (timedOut && timed))  
    22. 

  22.                 break;  
    23. 

  23.             if (compareAndDecrementWorkerCount(c))  
    24. 

  24.                 return null;  
    25. 

  25.             c = ctl.get();  // Re-read ctl  
    26. 

  26.             if (runStateOf(c) != rs)  
    27. 

  27.                 continue retry;  
    28. 

  28.             // else CAS failed due to workerCount change; retry inner loop  
    29. 

  29.         }  
    30. 

  30.   
    31. 

  31.         try {  
    32. 

  32.             Runnable r = timed ?  
    33. 

  33.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :  
    34. 

  34.                 workQueue.take();  
    35. 

  35.             if (r != null)  
    36. 

  36.                 return r;  
    37. 

  37.             timedOut = true;  
    38. 

  38.         } catch (InterruptedException retry) {  
    39. 

  39.             timedOut = false;  
    40. 

  40.         }  
    41. 

  41.     }  
    42. 

  42. }  
    43. 





若进入这个方法的工作线程是即将要终止的线程，该方法就必须返回null，有以下几种情形需要返回null：
1、当前活动线程数超过maximumPoolSize个（调用了setMaximumPoolSize的缘故）；
2、线程池已经停止（STOP）；
3、线程池已经关闭（SHUTDOWN）且任务队列为空；
4、工作线程获取任务超时，且满足(allowCoreThreadTimeOut || workerCount > corePoolSize)条件


先获取线程池的状态，如果状态大于等于STOP，也就是STOP、TIDYING、TERMINATED之一，这时候不管队列中有没有任务，都不用去执行了；如果线程池的状态为SHUTDOWN且队列中没有任务了，也不用继续执行了；所以这两种场景中获取任务的线程没必要存在了，这里调用了decrementWorkerCount减少活动线程数。前面在processWorkerExit中也提到，如果任务是非正常终止，processWorkerExit里要将活动线程数减一，正常的线程退出，减一是在这里做的。返回null之后，runWorker的while循环就退出了。接下来是个嵌套循环，它的目的就是上述的1和4.后面是从队列中取任务，比较简单，不多说。


以上，核心方法分析结束。欢迎指出理解错漏的地方。
															

																					

								

									相关推荐