【线程池】ThreadPoolExecutor源码分析 -- execute、shutdown方法

忘是亡心i 2023-06-08 04:48 20阅读 0赞

#### 简单介绍 ####

1.  来个简单的例子

ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10,100, 
        TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10));
      executor.execute(() -> {
          System.out.println("hello world");
      });

ThreadPoolExecutor构造方法至少需要5个参数：corePoolSize（线程池的核心线程数），maximumPoolSize（线程池最大线程数），keepAliveTime（线程空闲存活时间），TimeUnit（存活时间的单位），BlockingQueue（阻塞队列，用于存放Runnable）；  
另外两个参数ThreadFactory（线程工厂，用于创建新线程），RejectedExecutionHandler（拒绝策略），不传得话就会使用默认值。

1.  首先来看ThreadPoolExecutor的继承关系

![在这里插入图片描述][20191016145525887.png]  
3. 位于顶端的Executor接口很简单

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

1.  ExecutorService增加了一些方法  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70]  
    5.AbstractExecutorService实现了一些方法  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70 1]
2.  开始分析ThreadPoolExecutor源码，先来预先了解一些比较重要的成员变量、方法

//记录是运行状态和线程数量
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        private static final int COUNT_BITS = Integer.SIZE - 3;
        //线程允许的最大数量
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
        //运行状态：可以接受新任务，并执行已经加入任务队列的任务
        private static final int RUNNING    = -1 << COUNT_BITS;
        //关闭状态：不接受新任务，但仍然执行已经加入任务队列的任务
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        //停止状态：不接受新任务，不执行已经加入任务队列的任务，还中断正在执行的任务
        private static final int STOP       =  1 << COUNT_BITS;
        //调整回收状态：所有任务被停止，清空所有线程进入该状态，然后执行terminated()方法
        private static final int TIDYING    =  2 << COUNT_BITS;
        //终止状态：terminated()方法执行完成后进入该状态
        private static final int TERMINATED =  3 << COUNT_BITS;
    
        //获取线程池运行状态（即取高三位）
        private static int runStateOf(int c)     {  return c & ~CAPACITY; }
        private static int workerCountOf(int c)  {  return c & CAPACITY; }
        private static int ctlOf(int rs, int wc) {  return rs | wc; }

上面可以看到，线程池的运行状态值和当前线程数量都存在了一个int类型数字（ctl）里，高三位存的是状态，其他位用于记录数量。而且，状态是只能按照上面从上到下的顺序变化的。

##### 一、分析线程池执行过程 #####

1.  接着从execute()方法入手

public void execute(Runnable command) { 
        if (command == null)//判空
            throw new NullPointerException();
        int c = ctl.get();//拿到线程池状态
        if (workerCountOf(c) < corePoolSize) { 
    //核心线程数没达到，添加一个核心线程
            if (addWorker(command, true))//成功就返回，否则继续下面
                return;
    //要么是当前线程想添加一个核心线程的时候，核心线程数已经达到了；要么是线程池状态的原因，具体看下面addWorker方法
            c = ctl.get();//拿最新的状态值
        }
        if (isRunning(c) && workQueue.offer(command)) { //这里加入任务队列
            int recheck = ctl.get();
    //上面加入队列成功后进行二次检查
            if (!isRunning(recheck) && remove(command))
                reject(command);//线程池处于不接受新任务状态，调用拒绝策略
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);//线程池的无运行的线程，添加一个非核心线程
        }
        else if (!addWorker(command, false))//尝试再添加一个非核心线程来执行任务
            reject(command);//失败了，说明线程池状态原因或者队列满了
    }

1.  下面看addWorker方法是如何添加线程的

//第二个参数表示是否是核心线程，返回值表示是否添加线程成功
    private boolean addWorker(Runnable firstTask, boolean core) { 
       retry:
       for (;;) { 
           int c = ctl.get();//拿到线程池的状态
           int rs = runStateOf(c);//拿到运行状态
           if (rs >= SHUTDOWN &&
               ! (rs == SHUTDOWN &&
                  firstTask == null &&
                  ! workQueue.isEmpty()))
               return false;
    //这里主要是理解! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty())，
    //前面说过： SHUTDOWN状态不接受新任务，但仍然执行已经加入任务队列的任务，
    //所以当进入SHUTDOWN状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的，把这个条件取反就不允许了
    
           for (;;) { 
               int wc = workerCountOf(c);//拿到线程数
               if (wc >= CAPACITY ||
                   wc >= (core ? corePoolSize : maximumPoolSize))
                   return false;//大于最大容量，或大于初始参数值返回false
               if (compareAndIncrementWorkerCount(c))
                   break retry;//CAS添加线程数目成功，跳出最外层循环
               c = ctl.get();  // 重新拿到状态
               if (runStateOf(c) != rs)
                   continue retry;//如果运行状态变了，进入外层循环
               //否则继续在里层循环尝试CAS
           }
       }
    
    //上面添加线程数量成功了，开始真正添加线程
    
       boolean workerStarted = false;
       boolean workerAdded = false;
       Worker w = null;
       try { 
           w = new Worker(firstTask);//下面会分析Work类
           final Thread t = w.thread;
           if (t != null) { 
               final ReentrantLock mainLock = this.mainLock;
               mainLock.lock();
               try { 
                   int rs = runStateOf(ctl.get());
                   if (rs < SHUTDOWN ||
                       (rs == SHUTDOWN && firstTask == null)) { 
    //只有当前是正在运行状态，或是SHUTDOWN且firstTask为空，才进入
                       if (t.isAlive()) //这里说明被调了start方法
                           throw new IllegalThreadStateException();
                       workers.add(w);//添加线程,workers是个HashSet，所以要加锁保证线程安全
                       int s = workers.size();
                       if (s > largestPoolSize)
                           largestPoolSize = s;//记录一下线程池的峰值
                       workerAdded = true;//添加线程成功，设置标志位
                   }
               } finally { 
                   mainLock.unlock();
               }
    【标志位1】
               if (workerAdded) { 
                   t.start();//添加线程成功，开始运行线程
                   workerStarted = true;//设置开始运行标志位
               }
           }
       } finally { 
           if (!workerStarted)
               addWorkerFailed(w);//这里线程添加失败
       }
       return workerStarted;
    }

1.  从上面可以大致猜测出，在这个线程池的设计中，线程被封装成Worker的形式存在。  
    下面分析一下Worker类  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70 2]  
    AbstractQueuedSynchronizer在前面[ReentrantLock笔记(二) – 公平/非公平锁源码分析有分析过][ReentrantLock_ _]  
    一开始看到它还实现了Runnable接口觉得有点奇怪，不知道有什么用，接下去看就懂了。

final Thread thread;
    //初始化的Runnable
    Runnable firstTask;
    //完成的任务数
    volatile long completedTasks;
    Worker(Runnable firstTask) { 
        setState(-1); 
        this.firstTask = firstTask;
    //这里新建一个线程，注意传入的是this,这个很关键
        this.thread = getThreadFactory().newThread(this);
    }
    public void run() { 
        runWorker(this);
    }

可以看到该类有一个线程成员，而这个线程的runnable却是它自身，再看看它实现的run方法里执行了runWorker方法。回顾一下上面addWorker方法的【标志位1】处，那里启动了线程，所以会执行run方法，所以最终会调用runWorker方法。

runWorker方法是属于ThreadPoolExecutor的方法

final void runWorker(Worker w) { 
      Thread wt = Thread.currentThread();
      Runnable task = w.firstTask;
      w.firstTask = null;
      w.unlock(); // allow interrupts
      boolean completedAbruptly = true;//标志是不是用户任务异常导致终止的
      try { 
    //这里通过循环，不断地取任务来执行，getTask是会阻塞的
          while (task != null || (task = getTask()) != null) { 
              w.lock();
    //前面说过，stop状态时不接受新任务，不执行已经加入任务队列的任务，还中断正在执行的任务
    //所以对于stop状态以上是要中断线程的
    //(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP)确保线程中断标志位为true且是stop状态以上，接着清除了中断标志
    //!wt.isInterrupted()则再一次检查保证线程需要设置中断标志位
              if ((runStateAtLeast(ctl.get(), STOP) ||
                   (Thread.interrupted() &&
                    runStateAtLeast(ctl.get(), STOP))) &&
                  !wt.isInterrupted())
                  wt.interrupt();
              try { 
                  beforeExecute(wt, task);//回调方法，给子类具体实现
                  Throwable thrown = null;
                  try { 
                      task.run();
                  } catch (RuntimeException x) { 
                      thrown = x; throw x;
                  } catch (Error x) { 
                      thrown = x; throw x;
                  } catch (Throwable x) { 
                      thrown = x; throw new Error(x);
                  } finally { 
                      afterExecute(task, thrown);//回调方法，给子类具体实现
                  }
              } finally { 
                  task = null;//置空，如果进入下一个循环可以继续取任务
                  w.completedTasks++;//完成数+1
                  w.unlock();
              }
          }
          completedAbruptly = false;//说明不是用户任务异常引起的
      } finally { 
          processWorkerExit(w, completedAbruptly);
      }
    }

到这里基本就知道线程池如何复用线程，来执行任务了；也知道它怎么控制线程最大数量，但还不知道如何控制在超过空闲时间时回收线程？答案就在getTask方法

下面的getTask方法有点重要，起到控制线程池线程数量的作用

private Runnable getTask() { 
          boolean timedOut = false; // 取任务是否超时
          for (;;) { 
              int c = ctl.get();
              int rs = runStateOf(c);
        //这个状态判断挺重要的，起到线程池关闭作用
              if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                  decrementWorkerCount();//线程数量减一
                  return null;//这里返回null，意味着一个线程会退出
              }
              int wc = workerCountOf(c);
        //这里可以看出核心线程在空闲的时候也是可以设置被回收的
              boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        //timed为true将要有时间限制地取任务
    
              if ((wc > maximumPoolSize || (timed && timedOut))
                  && (wc > 1 || workQueue.isEmpty())) { 
            //大于最大限制线程数或超过空闲时间，并且当前线程数大于1或队列为空
                  if (compareAndDecrementWorkerCount(c))
                      return null;//说明线程数减一成功，返回null，意味着一个线程会退出
                  continue;//上面线程数减一失败，说明线程数量已被抢先改变，继续循环，
              }
    
              try { 
                  Runnable r = timed ?
                      workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                      workQueue.take();
                  if (r != null)
                      return r;
                  timedOut = true;//用于下一次循环中
              } catch (InterruptedException retry) { 
                  timedOut = false;
              }
          }
      }

上面的workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)用于超时检查取任务，超过时间就会返回null,那timedOut就会变为true,进入下一次循环，然后检查是否可以减少线程数（timedOut就是其中一个条件），然后返回null就可以退出一个线程了；  
PS: 上面有个很重要的判断if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty()))，这个用于关闭线程池

#### 二、分析线程池关闭过程 ####

分析完线程池的执行流程，下面接着分析下线程池如何关闭，看shutdown方法

public void shutdown() { 
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try { 
            checkShutdownAccess();//【步骤1】对线程检查一下，是否有权限修改
            advanceRunState(SHUTDOWN);//【步骤2】改变线程池状态为SHUTDOWN
            interruptIdleWorkers();//【步骤3】中断所有线程
            onShutdown(); // 【步骤4】留给子类具体实现，如ScheduledThreadPoolExecutor
        } finally { 
            mainLock.unlock();
        }
        tryTerminate();【步骤5】
    }

上面分了五个步骤

【步骤2】：

//该方法改变线程池状态为SHUTDOWN或者STOP
    private void advanceRunState(int targetState) { 
        for (;;) { 
            int c = ctl.get();
            if (runStateAtLeast(c, targetState) ||
                ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
                break;
        }
    }

【步骤3】

private void interruptIdleWorkers() { 
         interruptIdleWorkers(false);
     }
    
    private void interruptIdleWorkers(boolean onlyOne) { 
         final ReentrantLock mainLock = this.mainLock;
         mainLock.lock();
         try { 
             for (Worker w : workers) { 
                 Thread t = w.thread;
                 if (!t.isInterrupted() && w.tryLock()) { 
                     try { 
                         t.interrupt();//设置中断标志
                     } catch (SecurityException ignore) { 
                     } finally { 
                         w.unlock();
                     }
                 }
                 if (onlyOne)
                     break;
             }
         } finally { 
             mainLock.unlock();
         }
     }

【步骤5】

final void tryTerminate() { 
         for (;;) { 
             int c = ctl.get();
             if (isRunning(c) ||
                 runStateAtLeast(c, TIDYING) ||
                 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                 return;//状态条件不满足
             if (workerCountOf(c) != 0) { 
                 interruptIdleWorkers(ONLY_ONE);//线程数不为0，终止一个线程
                 return;
             }
             final ReentrantLock mainLock = this.mainLock;
             mainLock.lock();
             try { 
                 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { 
                     try { 
                         terminated();//设置TIDYING状态成功，调用terminated方法，具体由子类实现
                     } finally { 
                         ctl.set(ctlOf(TERMINATED, 0));//进入TERMINATED状态，说明已经关闭
                         termination.signalAll();//唤醒
                     }
                     return;
                 }
             } finally { 
                 mainLock.unlock();
             }
    //接着循环
         }
     }

看到termination.signalAll()的时候，有点疑惑，查了一下源码中用到termination（一个ConditionObject实例）的地方

public boolean awaitTermination(long timeout, TimeUnit unit)
         throws InterruptedException { 
         long nanos = unit.toNanos(timeout);
         final ReentrantLock mainLock = this.mainLock;
         mainLock.lock();
         try { 
             for (;;) { 
                 if (runStateAtLeast(ctl.get(), TERMINATED))
                     return true;
                 if (nanos <= 0)
                     return false;
                 nanos = termination.awaitNanos(nanos);
             }
         } finally { 
             mainLock.unlock();
         }
     }

是个公开的方法，联想一下平常用法，该方法用于检测线程池的关闭

try{ 
            while(!executor.awaitTermination(500, TimeUnit.MILLISECONDS)) { 
                //
            }
        } 
        catch (InterruptedException e) { 
            //中断处理
        }

从上面看出，shutdown方法改变状态为SHUTDOWN,并在尝试给每个线程设置中断标志，接着结合getTask()方法返回null来停止移除线程，最后尝试终止线程池。

参考：JDK1.8源码

[20191016145525887.png]: /images/20230531/1590e13dc3ba4b20a571abd8a2375096.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70]: /images/20230531/9be3d5f475fd4b8fb31c2b4ff0e63de5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70 1]: /images/20230531/e6b2a3a35aec4fa9843e153512f4c0b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5neW9uZzAxMjQ1_size_16_color_FFFFFF_t_70 2]: /images/20230531/e979042059744b1b9db5d1aebd28baa0.png
[ReentrantLock_ _]: https://blog.csdn.net/seasonLai/article/details/82501906