Java线程池ThreadPoolExecutor源码分析

朱雀 2024-04-03 08:34 66阅读 0赞

#### 目录 ####

*  一、线程池源码
 *   *  1.ThreadPoolExecutor应用方式
     *  2.ThreadPoolExecutor核心参数
     *  3.ThreadPoolExecutor执行流程
     *  4.ThreadPoolExecutor线程池状态
     *   *  4.1 线程池中核心属性ctl
         *  4.2 线程池的状态变换
     *  5.execute方法
     *  6.addWoker方法
     *  7.Woker对象
     *  8.runWoker方法
     *  9.getTask方法
     *  10.processWokerExit方法分析

## 一、线程池源码 ##

### 1.ThreadPoolExecutor应用方式 ###

为什么要使用线程池  
JDK中已经提供了Executors，提供了很多封装好的线程池，为什么还要手动创建  
线程池使用很简单

*  只需要构建好ThreadPoolExecutor对象即可，传入指定的参数
 *  在执行Runnable任务时，可以直接调用execute方法执行
 *  在执行Callable任务时，需要有返回结果，直接调用submit即可

### 2.ThreadPoolExecutor核心参数 ###

> public ThreadPoolExecutor(int corePoolSize, //核心线程数  
> int maximumPoolSize, //最大线程数  
> long keepAliveTime, //非核心线程的空间时间  
> TimeUnit unit, //keepAliveTime的单位  
> BlockingQueue workQueue, //阻塞/工作队列  
> ThreadFactory threadFactory, //线程工程  
> RejectedExecutionHandler handler) //拒绝策略

### 3.ThreadPoolExecutor执行流程 ###

![在这里插入图片描述][a0a07ed1745645ce84b7c197fe4d3665.png]

### 4.ThreadPoolExecutor线程池状态 ###

#### 4.1 线程池中核心属性ctl ####

//ctl本质就是一个int类型的数值
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //Integer.SIZE =32 COUNT_BITS=29
    //ctl表述了两个状态:
    //1.表示线程池当前的状态(高三位)
    //2.标识线程池当前的工作线程个数(低三位)
    private static final int COUNT_BITS = Integer.SIZE - 3;
    //表示了工作线程的最大数量
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    //00100000 00000000 00000000 00000000
    //00011111 11111111 11111111 11111111
    
    //线程池的5中状态
    //RUNNING
    //11111111 11111111 11111111 11111111
    //11100000 00000000 00000000 00000000
    //111
    private static final int RUNNING    = -1 << COUNT_BITS;
    //000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    //001
    private static final int STOP       =  1 << COUNT_BITS;
    //010
    private static final int TIDYING    =  2 << COUNT_BITS;
    //011
    private static final int TERMINATED =  3 << COUNT_BITS;
    
    //计算出当前线程池的状态
    //00011111 11111111 11111111 11111111
    //11100000 00000000 00000000 00000000
    //000  SHUTDOWN状态
    private static int runStateOf(int c)     {
         return c & ~CAPACITY; }
    //计算当前线程中的工作线程个数
    private static int workerCountOf(int c)  {
         return c & CAPACITY; }

#### 4.2 线程池的状态变换 ####

![在这里插入图片描述][744ecd7aab584fd8944a555e5f101cd3.png]

### 5.execute方法 ###

第一点核心:通过execute方法，查看线程池的整体执行流程,以及一些避免并发情况的判断  
第二点核心:为什么线程池会添加一个空任务的非核心线程到线程池

public void execute(Runnable command) {
        
          //非空判断
          if (command == null)
              throw new NullPointerException();
          //获取ctl属性
          int c = ctl.get();
          //工作线程的个数 是否小于 核心线程数
          if (workerCountOf(c) < corePoolSize) {
        
              //通过add方法，添加核心线程数
              if (addWorker(command, true))
                  //添加核心线程成功，返回true，直接return
                  return;
              //如果在并发情况下，添加核心线程失败的线程，需要重新获取ctl属性
              c = ctl.get();
          }
          //创建核心线程失败
          //判断当前线程池状态是否是Running
          //如果是Running，执行offer方法将任务添加到工作队列
          if (isRunning(c) && workQueue.offer(command)) {
        
              //添加任务到工作队列
              //再次重新获取ctl
              int recheck = ctl.get();
              //判断线程是否是Running状态，如果不是Running状态，需要将任务从工作队列中移除
              if (! isRunning(recheck) && remove(command))
                  //执行拒绝策略(线程池状态不正确，执行拒绝策略！)
                  reject(command);
              //判断工作线程是否为0
              else if (workerCountOf(recheck) == 0)
                  //工作线程数为0，但是工作队列中有任务在排队
                  //添加一个空任务非核心线程，为了处理在工作队列中排队的任务
                  addWorker(null, false);
          }
          //任务添加到工作队列失败，添加非核心线程去执行当前任务
          else if (!addWorker(command, false))
              //添加非核心线程失败，执行reject拒绝策略
              reject(command);

### 6.addWoker方法 ###

//添加工作线程
    private boolean addWorker(Runnable firstTask, boolean core) {
        
        //对线程状态的判断，以及对工作线程数量的判断
        //外层for循环的标识
        retry:
        for (;;) {
        
            //获取ctl的值
            int c = ctl.get();
            //拿到线程池的状态
            int rs = runStateOf(c);
            //如果线程池状态不是RUNNING,再次做后续判断，判断当前任务是否可以不处理
            if (rs >= SHUTDOWN &&
                //线程池状态为SHUTDOWN,并且任务为空，并且工作队列不为空
                //如果同时满足了这三个要求，那就是要处理工作工作队列的当前任务
                ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
                //只要不是RUNNING状态，不处理新任务
                //如果SHUTDOWN状态，并且满足了之前addWorder(null,false),并且工作队列有任务，不能走到当前
                return false;
            for (;;) {
        
                //基于ctl返回工作线程数量
                int wc = workerCountOf(c);
                //判断工作线程是否大于最大值
                if (wc >= CAPACITY ||
                    //如果是核心线程，是否大于设置的corePoolSize，如果是非核心是否大于maximumPoolSize
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    //当前工作线程已经达到最大值了
                    return false;
                //以CAS的方式，对工作线程数+1，如果成功，直接跳出外层for循环
                if (compareAndIncrementWorkerCount(c))
                    //直接跳出外层for循环
                    break retry;
                //重新获取ctl值
                c = ctl.get();  
                //基于新获取的ctl拿到线程池，判断和之前的rs状态是否一致
                if (runStateOf(c) != rs)
                    //说明并发操作导致线程池状态变化，需要重新判断状态
                    continue retry;
        
            }
        }
    
            //添加工作线程，并启动线程
            //工作线程是否启动
            boolean workerStarted = false;
            //工作线程是否添加了
            boolean workerAdded = false;
            //Worker就是工作线程
            Worker w = null;
            try {
        
                //new Woeker构造工作线程，将任务扔给Worker中
                w = new Worker(firstTask);
                //拿到了Worker中绑定的Thread线程
                final Thread t = w.thread;
                //肯定不为null
                if (t != null) {
        
                    //加锁.....
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
        
                        //基于重新获取的ctl，拿到线程池的状态
                        int rs = runStateOf(ctl.get());
                        //如果满足线程状态为RUNNING，就添加工作线程
                        if (rs < SHUTDOWN ||
                            //如果线程池状态为SHUTDOWN，并且传入的任务为null
                            (rs == SHUTDOWN && firstTask == null)) {
        
                            //开始添加工作线程
                            //判断当前线程是否处于run状态
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                            //将构建好的Worker对象添加到workers
                            workers.add(w);
                            //获取工作线程个数
                            int s = workers.size();
                            //如果现在的工作线程数，大于历史最大的工程线程数，就重新赋值给largestPoolSize
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                                //将工作线程添加的标识设置为true
                                workerAdded = true;
                        }
                    } finally {
        
                        //释放锁
                        mainLock.unlock();
                    }
                    if (workerAdded) {
        
                        //只要添加工作线程成功，启动线程
                        t.start();
                        //将工作线程启动的标识设置为true
                        workerStarted = true;
                    }
                }
            } finally {
        
                //如果启动线程失败！
                if (! workerStarted)
                    //移除wokers中的工作线程，将工作线程数-1，尝试将线程池状态变为TYDING
                    addWorkerFailed(w);
            }
            return workerStarted;
        }
    
    
    //启动工作线程失败后，做的补救操作
    private void addWorkerFailed(Worker w) {
        
       //加锁
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
        
           //判断之前创建工作线程是否成功
           if (w != null)
               //如果成功，就将workers中的当前工作线程移除
               workers.remove(w);
           //将工作线程数-1
           decrementWorkerCount();
           //尝试将线程池状态变为TYDING
           tryTerminate();
       } finally {
        
           mainLock.unlock();
       }
    }

### 7.Woker对象 ###

private final class Worker
            extends AbstractQueuedSynchronizer  //线程中断
            implements Runnable                 //存储需要执行的任务
        {
        
          
            private static final long serialVersionUID = 6138294804551838833L;
    
            //工作线程的Thread对象，初始化时候构建出来的
            final Thread thread;
            //需要执行的任务
            Runnable firstTask;
            volatile long completedTasks;
    
        
            Worker(Runnable firstTask) {
        
                //刚刚初始化的工作线程不允许被中断的
                setState(-1); 
                //第一次new的时候，会将任务赋值给firstTask
                this.firstTask = firstTask;
                //给Worker构建Thread对象
                this.thread = getThreadFactory().newThread(this);
            }
    
            //调用t.start()，执行当前的run方法
            public void run() {
        
                runWorker(this);
            }
    
            //中断线程不是立即让线程停止，只是将thread的中断的标识设置为true
    
            protected boolean isHeldExclusively() {
        
                return getState() != 0;
            }
            //将state置为0，在runWorker中，为了表示当前线程允许被中断
            protected boolean tryAcquire(int unused) {
        
                if (compareAndSetState(0, 1)) {
        
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            protected boolean tryRelease(int unused) {
        
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }

### 8.runWoker方法 ###

//执行任务...
    final void runWorker(Worker w) {
        
        //拿到工作线程
        Thread wt = Thread.currentThread();
        //拿到Worker对象中封装的任务
        Runnable task = w.firstTask;
        //将Worker的firstTask设置为null
        w.firstTask = null;
        //Worker的state归为0，代表可以被中断
        w.unlock(); 
        //执行任务时，勾子函数中是否出现异常的标识
        boolean completedAbruptly = true;
        try {
        
            //获取任务的第一个方式，就是执行execute，submit时(可能为false的情况是在execute中addworker了一个空任务的非核心线程为了解决队列)
            //获取任务的第二个方式,就是从工作队列中获取任务执行。
            while (task != null || (task = getTask()) != null) {
        
                //加锁,在SHUTDOWN状态下，当前线程不允许被中断
                //并且Worker内部实现的锁，并不是可重入锁，因为在中断时，也需要对worker进行lock，不能获取就代表当前工作线程正在执行任务
                w.lock();
                //如果线程池状态变为了STOP状态，必须将当前线程中断。
                //第一个判断:判断当前线程池状态是否是STOP
                //第二个判断:查看中断标记位，并归位，如果为false，说明不是STOP，如果变为true，需要再次查看是否是并发操作导致线程池为STOP
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    //查询当前线程中断标记是否为false，如果为false，就执行wt.interrupt();
                    !wt.isInterrupted())
                    //将中断标记设置为true
                    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
                    task = null;
                    w.completedTasks++;
                    //将state设置为0
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
        
            processWorkerExit(w, completedAbruptly);
        }
    }

### 9.getTask方法 ###

如何从工作队列获取任务

//从工作队列中获取任务
     private Runnable getTask() {
        
            //标识(非核心线程可以干掉)
            boolean timedOut = false; // Did the last poll() time out?
            //死循环
            for (;;) {
        
                //------------------------判断线程池状态------------------------
    
                //获取ctl标识
                int c = ctl.get();
                //拿到线程状态
                int rs = runStateOf(c);
                //如果进入if,需要干掉当前工作线程
                //线程池状态为SHUTDOWN、STOP
                //如果线程池状态大于等于STOP，需要移除掉当前工作线程
                //如果线程池状态为SHUTDOWN，并且工作队列为空，需要移除掉当前工作线程
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        
                    //移除当前工作线程
                    decrementWorkerCount();
                    //返回null，交给processworkerExit移除当前工作线程
                    return null;
                }
    
                //--------------------判断工作线程数量-----------------------------
                //拿到工作线程数
                int wc = workerCountOf(c);
    
                //allowCoreThreadTimeOut:是否允许核心线程超时(一般都为false)
                //工作线程是否大于核心线程
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
                if (
                     //工作线程是否已经大于最大线程数了
                //工作线程数大于核心线程数，并且当前线程已经超时
                //尝试干掉当前线程
                    (wc > maximumPoolSize || (timed && timedOut))
                    //工作线程数大于1，或者工作队列为空
                    //如果工作队列为空，我就干掉我自己
                    //如果工作线程数大于1，我就干掉我自己
                    && (wc > 1 || workQueue.isEmpty())) {
        
                    //基于CAS的方式移除掉当前线程，只有一个线程会CAS成功
                    if (compareAndDecrementWorkerCount(c))
                        //返回null，交给processworkerExit移除当前工作线程
                        return 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;
                }
            }
        }

### 10.processWokerExit方法分析 ###

//移除当前工作线程的操作
    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        
            //如果执行processworkerExit方法的操作不是getTask中的操作，是值接因为异常引起的。
            if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
                //因为执行方式"不合法",，手动扣减工作线程数
                decrementWorkerCount();
            final ReentrantLock mainLock = this.mainLock;
            //加锁！
            mainLock.lock();
            try {
        
                //记录当前线程池一共处理了多少个任务
                completedTaskCount += w.completedTasks;
                //移除工作线程
                workers.remove(w);
            } finally {
        
                mainLock.unlock();
            }
    
            //尝试将线程池关系(到过度状态 - 销毁状态)
            tryTerminate();
    
            //重新获取ctl
            int c = ctl.get();
            //当前线程池状态，进到这，说明是RUNNING、SHUTDOWN
            //不是STOP状态
            if (runStateLessThan(c, STOP)) {
        
                //如果是正常状态移除当前工作线程
                if (!completedAbruptly) {
        
                    //核心线程数最小值,允许多少
                    int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                    //如果工作队列中的任务不为空，设置工作线程的最小值为1
                    if (min == 0 && ! workQueue.isEmpty())
                        min = 1;
                    //还有工作线程在线程池中。
                    if (workerCountOf(c) >= min)
                        return; // replacement not needed
                }
                //说明是不正常的方式移除了当前工作线程，再添加一个工作线程
                //线程池工作队列不为空，并且没有工作线程，再添加一个工作线程
                addWorker(null, false);
            }
        }

[a0a07ed1745645ce84b7c197fe4d3665.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/03/0c807a8242aa47fca181a75c826bca2b.png
[744ecd7aab584fd8944a555e5f101cd3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/03/4786186575f7484e9c409052ff954259.png