ThreadPoolExecutor源码剖析

今天这篇文章来详细剖析一下ThreadPoolExecutor源码。

1.核心参数

    public ThreadPoolExecutor(int corePoolSize,//核心线程数限制
                              int maximumPoolSize,//最大线程限制
                              long keepAliveTime,//空闲线程存活时间
                              TimeUnit unit,//时间单位 seconds nano..
                              BlockingQueue<Runnable> workQueue,//任务队列
                              ThreadFactory threadFactory,//线程工厂
                              RejectedExecutionHandler handler/*拒绝策略*/)
    public void allowCoreThreadTimeOut(boolean value) {
        if (value && keepAliveTime <= 0)
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        if (value != allowCoreThreadTimeOut) {
            allowCoreThreadTimeOut = value;
            if (value)
                interruptIdleWorkers();
        }
    }

核心参数有如下几个:

  • corePoolSize 核心线程数限制
  • maximumPoolSize 最大线程限制
  • keepAliveTime 空闲线程存活时间
  • unit 时间单位
  • workQueue 任务队列
  • threadFactory 线程工厂
  • handler 拒绝策略
  • allowCoreThreadTimeOut 是否允许核心线程退出

2.核心属性以及线程池运行状态

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

核心就是这个原子类型的ctl,32位是个复合状态,高3位表示线程池运行状态,低29位表示线程池中所拥有的线程数量。

有一系列的位操作方法,就是获取操作线程池状态以及增减线程池中的线程数量。

    // Packing and unpacking ctl
    //获取当前线程池运行状态
    //~000 11111111111111111111 => 111 000000000000000000000
    //c == ctl = 111 000000000000000000111
    //111 000000000000000000111
    //111 000000000000000000000
    //111 000000000000000000000
    private static int runStateOf(int c)     { return c & ~CAPACITY; }

    //获取当前线程池线程数量
    //c == ctl = 111 000000000000000000111
    //111 000000000000000000111
    //000 111111111111111111111
    //000 000000000000000000111 => 7
    private static int workerCountOf(int c)  { return c & CAPACITY; }

接下来看看线程池状态:

    //111 000000000000000000  转换成整数,其实是一个负数
    private static final int RUNNING    = -1 << COUNT_BITS;
    //000 000000000000000000
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    //001 000000000000000000
    private static final int STOP       =  1 << COUNT_BITS;
    //010 000000000000000000
    private static final int TIDYING    =  2 << COUNT_BITS;
    //011 000000000000000000
    private static final int TERMINATED =  3 << COUNT_BITS;

线程池状态转移有两个核心方法:

2.1 优雅关闭shutdown方法

第一个就是优雅关闭的shutdown方法:

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        //获取线程池全局锁
        mainLock.lock();
        try {
            checkShutdownAccess();
            //设置线程池状态为SHUTDOWN
            advanceRunState(SHUTDOWN);
            //中断空闲线程
            interruptIdleWorkers();
            //空方法,子类可以扩展
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            //释放线程池全局锁
            mainLock.unlock();
        }
        //回头说..
        tryTerminate();
    }
    private void interruptIdleWorkers() {
        interruptIdleWorkers(false);
    }
    //onlyOne == true 说明只中断一个线程 ,false 则中断所有线程
    //共同前提,worker是空闲状态。
    private void interruptIdleWorkers(boolean onlyOne) {
        final ReentrantLock mainLock = this.mainLock;
        //持有全局锁
        mainLock.lock();
        try {
            //迭代所有worker
            for (Worker w : workers) {
                //获取当前worker的线程 保存到t
                Thread t = w.thread;
                //条件一:条件成立:!t.isInterrupted()  == true  说明当前迭代的这个线程尚未中断。
                //条件二:w.tryLock() 条件成立:说明当前worker处于空闲状态,可以去给它一个中断信号。 目前worker内的线程 在 queue.take() | queue.poll()
                //阻塞中。因为worker执行task时,是加锁的!
                if (!t.isInterrupted() && w.tryLock()) {
                    try {
                        //给当前线程中断信号..处于queue阻塞的线程,会被唤醒,唤醒后,进入下一次自旋时,可能会return null。执行退出相关的逻辑。
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    } finally {
                        //释放worker的独占锁。
                        w.unlock();
                    }
                }
                if (onlyOne)
                    break;
            }

        } finally {
            //释放全局锁。
            mainLock.unlock();
        }
    }

    final void tryTerminate() {
        //自旋
        for (;;) {
            //获取最新ctl值
            int c = ctl.get();
            //条件一:isRunning(c)  成立,直接返回就行,线程池很正常!
            //条件二:runStateAtLeast(c, TIDYING) 说明 已经有其它线程 在执行 TIDYING -> TERMINATED状态了,当前线程直接回去。
            //条件三:(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())
            //SHUTDOWN特殊情况,如果是这种情况,直接回去。得等队列中的任务处理完毕后,再转化状态。
            if (isRunning(c) ||
                    runStateAtLeast(c, TIDYING) ||
                    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;

            //什么情况会执行到这里?
            //1.线程池状态 >= STOP
            //2.线程池状态为 SHUTDOWN 且 队列已经空了

            //条件成立:当前线程池中的线程数量 > 0
            if (workerCountOf(c) != 0) { // Eligible to terminate
                //中断一个空闲线程。
                //空闲线程,在哪空闲呢? queue.take() | queue.poll()
                //1.唤醒后的线程 会在getTask()方法返回null
                //2.执行退出逻辑的时候会再次调用tryTerminate() 唤醒下一个空闲线程
                //3.因为线程池状态是 (线程池状态 >= STOP || 线程池状态为 SHUTDOWN 且 队列已经空了) 最终调用addWorker时,会失败。
                //最终空闲线程都会在这里退出,非空闲线程 当执行完当前task时,也会调用tryTerminate方法,有可能会走到这里。
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            //执行到这里的线程是谁?
            //workerCountOf(c) == 0 时,会来到这里。
            //最后一个退出的线程。 咱们知道,在 (线程池状态 >= STOP || 线程池状态为 SHUTDOWN 且 队列已经空了)
            //线程唤醒后,都会执行退出逻辑,退出过程中 会 先将 workerCount计数 -1 => ctl -1。
            //调用tryTerminate 方法之前,已经减过了,所以0时,表示这是最后一个退出的线程了。

            final ReentrantLock mainLock = this.mainLock;
            //获取线程池全局锁
            mainLock.lock();
            try {
                //设置线程池状态为TIDYING状态。
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        //调用钩子方法
                        terminated();
                    } finally {
                        //设置线程池状态为TERMINATED状态。
                        ctl.set(ctlOf(TERMINATED, 0));
                        //唤醒调用 awaitTermination() 方法的线程。
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                //释放线程池全局锁。
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }

如上两个方法都会尝试对空闲线程中断,空闲线程,在哪空闲呢? 一般是在runWorker()的getTask()里面阻塞在queue.take() | queue.poll()。

    //w 就是启动worker
    final void runWorker(Worker w) {
        //wt == w.thread
        Thread wt = Thread.currentThread();
        //将初始执行task赋值给task
        Runnable task = w.firstTask;
        //清空当前w.firstTask引用
        w.firstTask = null;
        //这里为什么先调用unlock? 就是为了初始化worker state == 0 和 exclusiveOwnerThread ==null
        w.unlock(); // allow interrupts

        //是否是突然退出,true->发生异常了,当前线程是突然退出,回头需要做一些处理
        //false->正常退出。
        boolean completedAbruptly = true;

        try {
            //条件一:task != null 指的就是firstTask是不是null,如果不是null,直接执行循环体里面。
            //条件二:(task = getTask()) != null   条件成立:说明当前线程在queue中获取任务成功,getTask这个方法是一个会阻塞线程的方法
            //getTask如果返回null,当前线程需要执行结束逻辑。
            while (task != null || (task = getTask()) != null) {
                //worker设置独占锁 为当前线程
                //为什么要设置独占锁呢?shutdown时会判断当前worker状态,根据独占锁是否空闲来判断当前worker是否正在工作。
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt

                //条件一:runStateAtLeast(ctl.get(), STOP)  说明线程池目前处于STOP/TIDYING/TERMINATION 此时线程一定要给它一个中断信号
                //条件一成立:runStateAtLeast(ctl.get(), STOP)&& !wt.isInterrupted()
                //上面如果成立:说明当前线程池状态是>=STOP 且 当前线程是未设置中断状态的,此时需要进入到if里面,给当前线程一个中断。

                //假设:runStateAtLeast(ctl.get(), STOP) == false
                // (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP)) 在干吗呢?
                // Thread.interrupted() 获取当前中断状态,且设置中断位为false。连续调用两次,这个interrupted()方法 第二次一定是返回false.
                // runStateAtLeast(ctl.get(), STOP) 大概率这里还是false.
                // 其实它在强制刷新当前线程的中断标记位 false,因为有可能上一次执行task时,业务代码里面将当前线程的中断标记位 设置为了 true,且没有处理
                // 这里一定要强制刷新一下。不会再影响到后面的task了。
                //假设:Thread.interrupted() == true  且 runStateAtLeast(ctl.get(), STOP)) == true
                //这种情况有发生几率么?
                //有可能,因为外部线程在 第一次 (runStateAtLeast(ctl.get(), STOP) == false 后,有机会调用shutdown 、shutdownNow方法,将线程池状态修改
                //这个时候,也会将当前线程的中断标记位 再次设置回 中断状态。
                if ((runStateAtLeast(ctl.get(), STOP) ||
                        (Thread.interrupted() &&
                                runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                    wt.interrupt();


                try {
                    //钩子方法,留给子类实现的
                    beforeExecute(wt, task);
                    //表示异常情况,如果thrown不为空,表示 task运行过程中 向上层抛出异常了。
                    Throwable thrown = null;
                    try {
                        //task 可能是FutureTask 也可能是 普通的Runnable接口实现类。
                        //如果前面是通过submit()提交的 runnable/callable 会被封装成 FutureTask。这个不清楚,请看上一期,在b站。
                        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;
                    //更新worker完成任务数量
                    w.completedTasks++;

                    //worker处理完一个任务后,会释放掉独占锁
                    //1.正常情况下,会再次回到getTask()那里获取任务  while(getTask...)
                    //2.task.run()时内部抛出异常了..
                    w.unlock();
                }
            }

            //什么情况下,会来到这里?
            //getTask()方法返回null时,说明当前线程应该执行退出逻辑了。
            completedAbruptly = false;
        } finally {

            //task.run()内部抛出异常时,直接从 w.unlock() 那里 跳到这一行。
            //正常退出 completedAbruptly == false
            //异常退出 completedAbruptly == true
            processWorkerExit(w, completedAbruptly);
        }
    }

来看看getTask()被中断情况下会返回啥?

    //什么情况下会返回null?
    //1.rs >= STOP 成立说明:当前的状态最低也是STOP状态,一定要返回null了
    //2.前置条件 状态是 SHUTDOWN ,workQueue.isEmpty()
    //3.线程池中的线程数量 超过 最大限制时,会有一部分线程返回Null
    //4.线程池中的线程数超过corePoolSize时,会有一部分线程 超时后,返回null。
    private Runnable getTask() {
        //表示当前线程获取任务是否超时 默认false true表示已超时
        boolean timedOut = false; // Did the last poll() time out?

        //自旋
        for (;;) {
            //获取最新ctl值保存到c中。
            int c = ctl.get();
            //获取线程池当前运行状态
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //条件一:rs >= SHUTDOWN 条件成立:说明当前线程池是非RUNNING状态,可能是 SHUTDOWN/STOP....
            //条件二:(rs >= STOP || workQueue.isEmpty())
            //2.1:rs >= STOP 成立说明:当前的状态最低也是STOP状态,一定要返回null了
            //2.2:前置条件 状态是 SHUTDOWN ,workQueue.isEmpty()条件成立:说明当前线程池状态为SHUTDOWN状态 且 任务队列已空,此时一定返回null。
            //返回null,runWorker方法就会将返回Null的线程执行线程退出线程池的逻辑。
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                //使用CAS+死循环的方式让 ctl值 -1
                decrementWorkerCount();
                return null;
            }

            //执行到这里,有几种情况?
            //1.线程池是RUNNING状态
            //2.线程池是SHUTDOWN状态 但是队列还未空,此时可以创建线程。

            //获取线程池中的线程数量
            int wc = workerCountOf(c);

            // Are workers subject to culling?
            //timed == true 表示当前这个线程 获取 task 时 是支持超时机制的,使用queue.poll(xxx,xxx); 当获取task超时的情况下,下一次自旋就可能返回null了。
            //timed == false 表示当前这个线程 获取 task 时 是不支持超时机制的,当前线程会使用 queue.take();

            //情况1:allowCoreThreadTimeOut == true 表示核心线程数量内的线程 也可以被回收。
            //所有线程 都是使用queue.poll(xxx,xxx) 超时机制这种方式获取task.
            //情况2:allowCoreThreadTimeOut == false 表示当前线程池会维护核心数量内的线程。
            //wc > corePoolSize
            //条件成立:当前线程池中的线程数量是大于核心线程数的,此时让所有路过这里的线程,都是用poll 支持超时的方式去获取任务,
            //这样,就会可能有一部分线程获取不到任务,获取不到任务 返回Null,然后..runWorker会执行线程退出逻辑。
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;


            //条件一:(wc > maximumPoolSize || (timed && timedOut))
            //1.1:wc > maximumPoolSize  为什么会成立?setMaximumPoolSize()方法,可能外部线程将线程池最大线程数设置为比初始化时的要小
            //1.2: (timed && timedOut) 条件成立:前置条件,当前线程使用 poll方式获取task。上一次循环时  使用poll方式获取任务时,超时了
            //条件一 为true 表示 线程可以被回收,达到回收标准,当确实需要回收时再回收。

            //条件二:(wc > 1 || workQueue.isEmpty())
            //2.1: wc > 1  条件成立,说明当前线程池中还有其他线程,当前线程可以直接回收,返回null
            //2.2: workQueue.isEmpty() 前置条件 wc == 1, 条件成立:说明当前任务队列 已经空了,最后一个线程,也可以放心的退出。
            if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
                //使用CAS机制 将 ctl值 -1 ,减1成功的线程,返回null
                //CAS成功的,返回Null
                //CAS失败? 为什么会CAS失败?
                //1.其它线程先你一步退出了
                //2.线程池状态发生变化了。
                if (compareAndDecrementWorkerCount(c))
                    return null;
                //再次自旋时,timed有可能就是false了,因为当前线程cas失败,很有可能是因为其它线程成功退出导致的,再次咨询时
                //检查发现,当前线程 就可能属于 不需要回收范围内了。
                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为空,并且此时线程池状态已经改为了SHUTDOWN,这是肯定会返回null,返回null以后,runWorker()方法就会执行线程退出逻辑processWorkerExit(w, completedAbruptly)

    private void processWorkerExit(Worker w, boolean completedAbruptly) {
        //条件成立:代表当前w 这个worker是发生异常退出的,task任务执行过程中向上抛出异常了..
        //异常退出时,ctl计数,并没有-1
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();

        //获取线程池的全局锁引用
        final ReentrantLock mainLock = this.mainLock;
        //加锁
        mainLock.lock();
        try {
            //将当前worker完成的task数量,汇总到线程池的completedTaskCount
            completedTaskCount += w.completedTasks;
            //将worker从池子中移除..
            workers.remove(w);
        } finally {
            //释放全局锁
            mainLock.unlock();
        }


        tryTerminate();

        //获取最新ctl值
        int c = ctl.get();
        //条件成立:当前线程池状态为 RUNNING 或者 SHUTDOWN状态
        if (runStateLessThan(c, STOP)) {

            //条件成立:当前线程是正常退出..
            if (!completedAbruptly) {

                //min表示线程池最低持有的线程数量
                //allowCoreThreadTimeOut == true => 说明核心线程数内的线程,也会超时被回收。 min == 0
                //allowCoreThreadTimeOut == false => min == corePoolSize
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;


                //线程池状态:RUNNING SHUTDOWN
                //条件一:假设min == 0 成立
                //条件二:! workQueue.isEmpty() 说明任务队列中还有任务,最起码要留一个线程。
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;

                //条件成立:线程池中还拥有足够的线程。
                //考虑一个问题: workerCountOf(c) >= min  =>  (0 >= 0) ?
                //有可能!
                //什么情况下? 当线程池中的核心线程数是可以被回收的情况下,会出现这种情况,这种情况下,当前线程池中的线程数 会变为0
                //下次再提交任务时,会再创建线程。
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }

            //1.当前线程在执行task时 发生异常,这里一定要创建一个新worker顶上去。
            //2.!workQueue.isEmpty() 说明任务队列中还有任务,最起码要留一个线程。 当前状态为 RUNNING || SHUTDOWN
            //3.当前线程数量 < corePoolSize值,此时会创建线程,维护线程池数量在corePoolSize个。
            addWorker(null, false);
        }
    }

2.2暴力关闭线程池

shutdown():

把线程池的状态设置成SHUTDOWN状态,然后中断所有空闲的线程

shutdownNow():

首先把线程池的状态设置成STOP,然后尝试停止所有正在执行任务或者空闲的线程,并将任务队列清空,然后返回执行任务中还没有执行的任务。

    public List<Runnable> shutdownNow() {
        //返回值引用
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        //获取线程池全局锁
        mainLock.lock();
        try {
            checkShutdownAccess();
            //设置线程池状态为STOP
            advanceRunState(STOP);
            //中断线程池中所有线程
            interruptWorkers();
            //导出未处理的task
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }

        tryTerminate();
        //返回当前任务队列中 未处理的任务。
        return tasks;
    }
    private void interruptWorkers() {
        final ReentrantLock mainLock = this.mainLock;
        //获取线程池全局锁
        mainLock.lock();
        try {
            //遍历所有worker
            for (Worker w : workers)
                //interruptIfStarted() 如果worker内的thread 是启动状态,则给它一个中断信号。。
                w.interruptIfStarted();
        } finally {
            //释放线程池全局锁
            mainLock.unlock();
        }
    }

3.线程池的运行逻辑

    //command 可以是普通的Runnable 实现类,也可以是 FutureTask
    public void execute(Runnable command) {
        //非空判断..
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        //获取ctl最新值赋值给c,ctl :高3位 表示线程池状态,低位表示当前线程池线程数量。
        int c = ctl.get();
        //workerCountOf(c) 获取出当前线程数量
        //条件成立:表示当前线程数量小于核心线程数,此次提交任务,直接创建一个新的worker,对应线程池中多了一个新的线程。
        if (workerCountOf(c) < corePoolSize) {
            //addWorker 即为创建线程的过程,会创建worker对象,并且将command作为firstTask
            //core == true 表示采用核心线程数量限制  false表示采用 maximumPoolSize
            if (addWorker(command, true))
                //创建成功后,直接返回。addWorker方法里面会启动新创建的worker,将firstTask执行。
                return;

            //执行到这条语句,说明addWorker一定是失败了...
            //有几种可能呢??
            //1.存在并发现象,execute方法是可能有多个线程同时调用的,当workerCountOf(c) < corePoolSize成立后,
            //其它线程可能也成立了,并且向线程池中创建了worker。这个时候线程池中的核心线程数已经达到,所以...
            //2.当前线程池状态发生改变了。 RUNNING SHUTDOWN STOP TIDYING TERMINATION
            //当线程池状态是非RUNNING状态时,addWorker(firstTask!=null, true|false) 一定会失败。
            //SHUTDOWN 状态下,也有可能创建成功。前提 firstTask == null 而且当前 queue  不为空。特殊情况。
            c = ctl.get();
        }


        //执行到这里有几种情况?
        //1.当前线程数量已经达到corePoolSize
        //2.addWorker失败..

        //条件成立:说明当前线程池处于running状态,则尝试将 task 放入到workQueue中。
        if (isRunning(c) && workQueue.offer(command)) {
            //执行到这里,说明offer提交任务成功了..

            //再次获取ctl保存到recheck。
            int recheck = ctl.get();

            //条件一:! isRunning(recheck) 成立:说明你提交到队列之后,线程池状态被外部线程给修改 比如:shutdown() shutdownNow()
            //这种情况 需要把刚刚提交的任务删除掉。
            //条件二:remove(command) 有可能成功,也有可能失败
            //成功:提交之后,线程池中的线程还未消费(处理)
            //失败:提交之后,在shutdown() shutdownNow()之前,就被线程池中的线程 给处理。
            if (! isRunning(recheck) && remove(command))
                //提交之后线程池状态为 非running 且 任务出队成功,走个拒绝策略。
                reject(command);

            //有几种情况会到这里?
            //1.当前线程池是running状态(这个概率最大)
            //2.线程池状态是非running状态 但是remove提交的任务失败.

            //担心 当前线程池是running状态,但是线程池中的存活线程数量是0,这个时候,如果是0的话,会很尴尬,任务没线程去跑了,
            //这里其实是一个担保机制,保证线程池在running状态下,最起码得有一个线程在工作。
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }

        //执行到这里,有几种情况?
        //1.offer失败
        //2.当前线程池是非running状态

        //1.offer失败,需要做什么? 说明当前queue 满了!这个时候 如果当前线程数量尚未达到maximumPoolSize的话,会创建新的worker直接执行command
        //假设当前线程数量达到maximumPoolSize的话,这里也会失败,也走拒绝策略。

        //2.线程池状态为非running状态,这个时候因为 command != null addWorker 一定是返回false。
        else if (!addWorker(command, false))
            reject(command);

    }
    //firstTask 可以为null,表示启动worker之后,worker自动到queue中获取任务.. 如果不是null,则worker优先执行firstTask
    //core 采用的线程数限制 如果为true 采用 核心线程数限制  false采用 maximumPoolSize线程数限制.

    //返回值总结:
    //true 表示创建worker成功,且线程启动

    //false 表示创建失败。
    //1.线程池状态rs > SHUTDOWN (STOP/TIDYING/TERMINATION)
    //2.rs == SHUTDOWN 但是队列中已经没有任务了 或者 当前状态是SHUTDOWN且队列未空,但是firstTask不为null
    //3.当前线程池已经达到指定指标(coprePoolSize 或者 maximumPoolSIze)
    //4.threadFactory 创建的线程是null
    private boolean addWorker(Runnable firstTask, boolean core) {
        //自旋 判断当前线程池状态是否允许创建线程的事情。
        retry:
        for (;;) {
            //获取当前ctl值保存到c
            int c = ctl.get();
            //获取当前线程池运行状态 保存到rs长
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.

            //条件一:rs >= SHUTDOWN 成立:说明当前线程池状态不是running状态
            //条件二:前置条件,当前的线程池状态不是running状态  ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())
            //rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()
            //表示:当前线程池状态是SHUTDOWN状态 & 提交的任务是空,addWorker这个方法可能不是execute调用的。 & 当前任务队列不是空
            //排除掉这种情况,当前线程池是SHUTDOWN状态,但是队列里面还有任务尚未处理完,这个时候是允许添加worker,但是不允许再次提交task。
            if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                            firstTask == null &&
                            ! workQueue.isEmpty()))
                //什么情况下回返回false?
                //线程池状态 rs > SHUTDOWN
                //rs == SHUTDOWN 但是队列中已经没有任务了 或者 rs == SHUTDOWN 且 firstTask != null
                return false;

            //上面这些代码,就是判断 当前线程池状态 是否允许添加线程。


            //内部自旋 获取创建线程令牌的过程。
            for (;;) {
                //获取当前线程池中线程数量 保存到wc中
                int wc = workerCountOf(c);

                //条件一:wc >= CAPACITY 永远不成立,因为CAPACITY是一个5亿多大的数字
                //条件二:wc >= (core ? corePoolSize : maximumPoolSize)
                //core == true ,判断当前线程数量是否>=corePoolSize,会拿核心线程数量做限制。
                //core == false,判断当前线程数量是否>=maximumPoolSize,会拿最大线程数量做限制。
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    //执行到这里,说明当前无法添加线程了,已经达到指定限制了
                    return false;

                //条件成立:说明记录线程数量已经加1成功,相当于申请到了一块令牌。
                //条件失败:说明可能有其它线程,修改过ctl这个值了。
                //可能发生过什么事?
                //1.其它线程execute() 申请过令牌了,在这之前。导致CAS失败
                //2.外部线程可能调用过 shutdown() 或者 shutdownNow() 导致线程池状态发生变化了,咱们知道 ctl 高3位表示状态
                //状态改变后,cas也会失败。
                if (compareAndIncrementWorkerCount(c))
                    //进入到这里面,一定是cas成功啦!申请到令牌了
                    //直接跳出了 retry 外部这个for自旋。
                    break retry;

                //CAS失败,没有成功的申请到令牌
                //获取最新的ctl值
                c = ctl.get();  // Re-read ctl
                //判断当前线程池状态是否发生过变化,如果外部在这之前调用过shutdown. shutdownNow 会导致状态变化。
                if (runStateOf(c) != rs)
                    //状态发生变化后,直接返回到外层循环,外层循环负责判断当前线程池状态,是否允许创建线程。
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }





        //表示创建的worker是否已经启动,false未启动  true启动
        boolean workerStarted = false;
        //表示创建的worker是否添加到池子中了 默认false 未添加 true是添加。
        boolean workerAdded = false;

        //w表示后面创建worker的一个引用。
        Worker w = null;
        try {
            //创建Worker,执行完后,线程应该是已经创建好了。
            w = new Worker(firstTask);

            //将新创建的worker节点的线程 赋值给 t
            final Thread t = w.thread;

            //为什么要做 t != null 这个判断?
            //为了防止ThreadFactory 实现类有bug,因为ThreadFactory 是一个接口,谁都可以实现。
            //万一哪个 小哥哥 脑子一热,有bug,创建出来的线程 是null、、
            //Doug lea考虑的比较全面。肯定会防止他自己的程序报空指针,所以这里一定要做!
            if (t != null) {
                //将全局锁的引用保存到mainLock
                final ReentrantLock mainLock = this.mainLock;
                //持有全局锁,可能会阻塞,直到获取成功为止,同一时刻 操纵 线程池内部相关的操作,都必须持锁。
                mainLock.lock();
                //从这里加锁之后,其它线程 是无法修改当前线程池状态的。
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    //获取最新线程池运行状态保存到rs中
                    int rs = runStateOf(ctl.get());

                    //条件一:rs < SHUTDOWN 成立:最正常状态,当前线程池为RUNNING状态.
                    //条件二:前置条件:当前线程池状态不是RUNNING状态。
                    //(rs == SHUTDOWN && firstTask == null)  当前状态为SHUTDOWN状态且firstTask为空。其实判断的就是SHUTDOWN状态下的特殊情况,
                    //只不过这里不再判断队列是否为空了
                    if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                        //t.isAlive() 当线程start后,线程isAlive会返回true。
                        //防止脑子发热的程序员,ThreadFactory创建线程返回给外部之前,将线程start了。。
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();

                        //将咱们创建的worker添加到线程池中。
                        workers.add(w);
                        //获取最新当前线程池线程数量
                        int s = workers.size();
                        //条件成立:说明当前线程数量是一个新高。更新largestPoolSize
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        //表示线程已经追加进线程池中了。
                        workerAdded = true;
                    }
                } finally {
                    //释放线程池全局锁。
                    mainLock.unlock();
                }
                //条件成立:说明 添加worker成功
                //条件失败:说明线程池在lock之前,线程池状态发生了变化,导致添加失败。
                if (workerAdded) {
                    //成功后,则将创建的worker启动,线程启动。
                    t.start();
                    //启动标记设置为true
                    workerStarted = true;
                }
            }

        } finally {
            //条件成立:! workerStarted 说明启动失败,需要做清理工作。
            if (! workerStarted)
                //失败时做什么清理工作?
                //1.释放令牌
                //2.将当前worker清理出workers集合
                addWorkerFailed(w);
        }

        //返回新创建的线程是否启动。
        return workerStarted;
    }

4.Worker的运行逻辑

在上面的addWorker里面首先是new Worker(),然后是t.start()启动Worker线程。

这里state为-1禁止中断的理解:

  • 在addWorker里面,首先new Worker()将state设置为-1,然后将worker加入到线程池集合里面去
    然后才是t.start() -> runWorker(),这里将state设置为0
    这中间是可以禁止对该worker进行中断的
    因为此时state为-1,在中断前,都会w.tryLock()肯定会失败,所以当然就禁止了中断
  • 在start之前,新创建的线程会加入到hashset workers里面去
    其他地方可以调用shutdown shutdownnow
    这些关闭方法都避开了初始状态没有start的线程
    不对这些线程中断
    可能是未启动的线程interrupt,中断语义会不正确,因为此时都没有真正的内核线程与之绑定
    所以才设置为-1,避免对其进行错误的中断
        //firstTask可以为null。为null 启动后会到queue中获取。
        Worker(Runnable firstTask) {
            //设置AQS独占模式为初始化中状态,这个时候 不能被抢占锁。
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            //使用线程工厂创建了一个线程,并且将当前worker 指定为 Runnable,也就是说当thread启动的时候,会以worker.run()为入口。
            this.thread = getThreadFactory().newThread(this);
        }

然后t.start()会调用至Worker.run()首先设置worker state == 0 和 exclusiveOwnerThread ==null

        //当worker启动时,会执行run()
        public void run() {
            //ThreadPoolExecutor->runWorker() 这个是核心方法,等后面分析worker启动后逻辑时会以这里切入。
            runWorker(this);
        }

runWorker里面首先将

    //w 就是启动worker
    final void runWorker(Worker w) {
        //wt == w.thread
        Thread wt = Thread.currentThread();
        //将初始执行task赋值给task
        Runnable task = w.firstTask;
        //清空当前w.firstTask引用
        w.firstTask = null;
        //这里为什么先调用unlock? 就是为了初始化worker state == 0 和 exclusiveOwnerThread ==null
        w.unlock(); // allow interrupts

        //是否是突然退出,true->发生异常了,当前线程是突然退出,回头需要做一些处理
        //false->正常退出。
        boolean completedAbruptly = true;

        try {
            //条件一:task != null 指的就是firstTask是不是null,如果不是null,直接执行循环体里面。
            //条件二:(task = getTask()) != null   条件成立:说明当前线程在queue中获取任务成功,getTask这个方法是一个会阻塞线程的方法
            //getTask如果返回null,当前线程需要执行结束逻辑。
            while (task != null || (task = getTask()) != null) {
                //worker设置独占锁 为当前线程
                //为什么要设置独占锁呢?shutdown时会判断当前worker状态,根据独占锁是否空闲来判断当前worker是否正在工作。
                w.lock();
                // If pool is stopping, ensure thread is interrupted;
                // if not, ensure thread is not interrupted.  This
                // requires a recheck in second case to deal with
                // shutdownNow race while clearing interrupt

                //条件一:runStateAtLeast(ctl.get(), STOP)  说明线程池目前处于STOP/TIDYING/TERMINATION 此时线程一定要给它一个中断信号
                //条件一成立:runStateAtLeast(ctl.get(), STOP)&& !wt.isInterrupted()
                //上面如果成立:说明当前线程池状态是>=STOP 且 当前线程是未设置中断状态的,此时需要进入到if里面,给当前线程一个中断。

                //假设:runStateAtLeast(ctl.get(), STOP) == false
                // (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP)) 在干吗呢?
                // Thread.interrupted() 获取当前中断状态,且设置中断位为false。连续调用两次,这个interrupted()方法 第二次一定是返回false.
                // runStateAtLeast(ctl.get(), STOP) 大概率这里还是false.
                // 其实它在强制刷新当前线程的中断标记位 false,因为有可能上一次执行task时,业务代码里面将当前线程的中断标记位 设置为了 true,且没有处理
                // 这里一定要强制刷新一下。不会再影响到后面的task了。
                //假设:Thread.interrupted() == true  且 runStateAtLeast(ctl.get(), STOP)) == true
                //这种情况有发生几率么?
                //有可能,因为外部线程在 第一次 (runStateAtLeast(ctl.get(), STOP) == false 后,有机会调用shutdown 、shutdownNow方法,将线程池状态修改
                //这个时候,也会将当前线程的中断标记位 再次设置回 中断状态。
                if ((runStateAtLeast(ctl.get(), STOP) ||
                        (Thread.interrupted() &&
                                runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                    wt.interrupt();


                try {
                    //钩子方法,留给子类实现的
                    beforeExecute(wt, task);
                    //表示异常情况,如果thrown不为空,表示 task运行过程中 向上层抛出异常了。
                    Throwable thrown = null;
                    try {
                        //task 可能是FutureTask 也可能是 普通的Runnable接口实现类。
                        //如果前面是通过submit()提交的 runnable/callable 会被封装成 FutureTask。这个不清楚,请看上一期,在b站。
                        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;
                    //更新worker完成任务数量
                    w.completedTasks++;

                    //worker处理完一个任务后,会释放掉独占锁
                    //1.正常情况下,会再次回到getTask()那里获取任务  while(getTask...)
                    //2.task.run()时内部抛出异常了..
                    w.unlock();
                }
            }

            //什么情况下,会来到这里?
            //getTask()方法返回null时,说明当前线程应该执行退出逻辑了。
            completedAbruptly = false;
        } finally {

            //task.run()内部抛出异常时,直接从 w.unlock() 那里 跳到这一行。
            //正常退出 completedAbruptly == false
            //异常退出 completedAbruptly == true
            processWorkerExit(w, completedAbruptly);
        }
    }
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,098评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,213评论 2 380
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 149,960评论 0 336
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,519评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,512评论 5 364
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,533评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,914评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,574评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,804评论 1 296
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,563评论 2 319
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,644评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,350评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,933评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,908评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,146评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,847评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,361评论 2 342

推荐阅读更多精彩内容