【ThreadPoolExecutor】深入浅出的核心源码分析｜周末学习

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.
         * 
         * 如果运行的线程数小于corePoolSize，尝试创建一个新线程（Worker），并执行     * 它的第一个任务command。
         * 
         * 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.
         * 
         *  如果task成功插入等待队列，我们仍需要进行双重校验是否可以成功添加一个线程
　　　　　 *　（因为有的线程可能在我们上次检查以后已经死掉了）或者在我们进入这个
　　　　　 *　 方法后线程池已经关闭了。
　　　　　 *　 
         * 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.
         * 
　　　　　      如果等待队列已满，我们尝试新创建一个线程。如果创建失败，我们知道线程已关闭
　　　　　      或者已饱和，因此我们拒绝改任务。
         */
        int c = ctl.get();
        // 高3位表示状态，低29位任务数量。
　　　　  //工作线程小于核心线程数，创建新的线程。
        if (workerCountOf(c) < corePoolSize) {
　　　　　　　 //创建新的worker立即执行command，并且轮训workQueue处理task。
　　　　　　　 （核心线程数）
            if (addWorker(command, true)) 
                return;
            c = ctl.get();
        }
　　　　　//线程池在运行状态且可以将task插入队列
　　　　　//第一次校验线程池在运行状态
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
　　　　　　　　//第二次校验，防止在第一次校验通过后线程池关闭。
　　　　　　　　如果线程池关闭，在队列中删除task并拒绝task
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //如果线程数=0（线程都死掉了，比如：corePoolSize=0），新建线程且未指
            // 定firstTask，仅仅去轮训workQueue
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
　　　　　//线程队列已满，尝试创建新线程执行task，创建失败后拒绝task
        //创建失败原因：1.线程池关闭；2.线程数已经达到maxPoolSize
        else if (!addWorker(command, false))
            reject(command);
    }
复制代码

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
    　　　　 //外层循环判断线程池的状态
         for (;;) {
            //在进行获取一次上下文操作机制
            int c = ctl.get();
            int rs = runStateOf(c);
             //线程池状态
            // Check if queue empty only if necessary.
    　　　　  // 线程池状态:RUNNING = -1、SHUTDOWN = 0、 STOP = 1、TIDYING = 2、TERMINATED = 3　
　　　　　　　　//线程池至少是shutdown状态
　　　　　　　 if (rs >= SHUTDOWN &&
　　　　　　　　　　// 除了线程池正在关闭（shutdown），
　　　　　　　　　　// 队列里还有未处理的task的情况，其他都不能添加
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            //内层循环判断是否到达容量上限，worker+1
            for (;;) {
                int wc = workerCountOf(c);//worker数量
                //worker大于Integer最大上限
                //或到达边界上限
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS worker+1
                if (compareAndIncrementWorkerCount(c))
                    break retry;//成功了跳出循环
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs) 
                //如果线程池状态发生变化，重试外层循环
                    continue retry;
                // else CAS failed due to workerCount change;
                // retry inner loop 
　　　　　　　　　// CAS失败workerCount被其他线程改变，
　　　　　　　　　// 重新尝试内层循环CAS对workerCount+1
            }
        }
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock;
            w = new Worker(firstTask); 
            //1.state置为-1，Worker继承了AbstractQueuedSynchronizer.
            //2.设置firstTask属性.
            //3.Worker实现了Runable接口，将this作为入参创建线程.
            final Thread t = w.thread;
            if (t != null) {
　　　　　　　　　　//addWorker需要加锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int c = ctl.get();
                    int rs = runStateOf(c);
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//workers是HashSet<Worker>
　　　　　　　　　　　　　　//设置最大线程池大小
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (!workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
复制代码

/** 
　　　* Class Worker mainly maintains interrupt control state for
     * threads running tasks, along with other minor bookkeeping.
     * This class opportunistically extends AbstractQueuedSynchronizer
     * to simplify acquiring and releasing a lock surrounding each
     * task execution.  This protects against interrupts that are
     * intended to wake up a worker thread waiting for a task from
     * instead interrupting a task being run.  We implement a simple
     * non-reentrant mutual exclusion lock rather than use
     * ReentrantLock because we do not want worker tasks to be able to
     * reacquire the lock when they invoke pool control methods like
     * setCorePoolSize.  Additionally, to suppress interrupts until
     * the thread actually starts running tasks, we initialize lock
     * state to a negative value, and clear it upon start (in
     * runWorker).
　　　*  1.Worker类主要负责运行线程状态的控制。
　　　*  2.Worker继承了AQS实现了简单的获取锁和释放所的操作。来避免中断等待执行任务的线
　　　*  程时，中断正在运行中的线程（线程刚启动，还没开始执行任务）。
　　　*  3.自己实现不可重入锁，是为了避免在实现线程池控状态控制的方法，例如：
　　　*  setCorePoolSize的时候中断正在开始运行的线程。
　　   *  setCorePoolSize可能会调用interruptIdleWorkers()，该方法中会调用worker的tryLock()方法
　　    *  中断线程，自己实现锁可以确保工作线程启动之前不会被中断
     */
    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        /**
         * This class will never be serialized, but we provide a
         * serialVersionUID to suppress a javac warning.
         */
        private static final long serialVersionUID = 6138294804551838833L;
        /** Thread this worker is running in.  Null if factory fails. */
         封装任务线程机制。
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        Runnable firstTask;
        /** Per-thread task counter */
        volatile long completedTasks;
        /**
         * Creates with given first task and thread from ThreadFactory.
         * @param firstTask the first task (null if none)
         */
        Worker(Runnable firstTask) {
        // inhibit interrupts until runWorker //状态置为-1，如果中断线程需要CAS将state 从0-
        >1，以此来保证能只中断从workerQueue getTask的线程
            setState(-1); 
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        // 核心方法机制
        /** Delegates main run loop to outer runWorker  */
        public void run() {
          //首先执行w.unlock,就是把state置为0，对该线程的中断就可以进行了
            runWorker(this);
        }
        // Lock methods
        //
        // The value 0 represents the unlocked state.
        // The value 1 represents the locked state.
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
        // 在setCorePoolSize/shutdown等方法中断worker线程时需要调用该方法，
        // 确保中断的是从workerQueue getTask的线程
        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;
        }
        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        //调用tryRelease修改state=0，LockSupport.unpark(thread) 下一个等待锁的线程
        public boolean isLocked() { return isHeldExclusively(); }
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }
复制代码