ThreadPoolExecutor学习笔记

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

 private static final int CAPACITY   = (1 << COUNT_BITS) - 1;//线程池最大线程数=536870911（2^29-1）

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

private static int ctlOf(int rs, int wc) { return rs | wc; }

/**
 * 这个方法用于取出runState的值 因为CAPACITY值为：00011111111111111111111111111111
 * ~为按位取反操作，则~CAPACITY值为：11100000000000000000000000000000
 * 再同参数做&操作，就将低29位置0了，而高3位还是保持原先的值，也就是runState的值
 * 
 * @param c 该参数为存储runState和workerCount的int值
 * @return runState的值
 */
private static int runStateOf(int c) {
    return c & ~CAPACITY;
}


/**
 * 这个方法用于取出workerCount的值
 * 因为CAPACITY值为：00011111111111111111111111111111，所以&操作将参数的高3位置0了
 * 保留参数的低29位，也就是workerCount的值
 * 
 * @param c  ctl, 存储runState和workerCount的int值
 * @return workerCount的值
 */
private static int workerCountOf(int c) {
    return c & CAPACITY;
}

/**
 * 将runState和workerCount存到同一个int中
 * “|”运算的意思是，假设rs的值是101000，wc的值是000111，则他们位或运算的值为101111
 * 
 * @param rs runState移位过后的值，负责填充返回值的高3位
 * @param wc  workerCount移位过后的值，负责填充返回值的低29位
 * @return 两者或运算过后的值
 */
private static int ctlOf(int rs, int wc) {
    return rs | wc;
}

// 只有RUNNING状态会小于0
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}

 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;//当前worker对象关联的线程
        /** 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) {
            //禁止中断直到执行了 runWorker 方法，在shutdownNow 中断线程之前，会首先判断 state 是否大于  等于 0
        // 所以这里将 state 设为 -1，可以防止当前线程被中断
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

        /** Delegates main run loop to outer runWorker  */
        public void run() {
            runWorker(this);
        }

        // Lock methods
        //
        // The value 0 represents the unlocked state.
        // The value 1 represents the locked state.

        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        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); }
        public boolean isLocked() { return isHeldExclusively(); }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

 public Future<?> submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }

  public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
 
        int c = ctl.get();
        // 线程数量少于 corePoolSize，将改任务分配给一个新建的线程
        if (workerCountOf(c) < corePoolSize) {
            // true 表示将当前线程添加为核心线程
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
      //线程数量大于等于 corePoolSize，首先尝试将任务添加到任务队列
      // workQueue.offer 会将任务添加到队列尾部
        if (isRunning(c) && workQueue.offer(command)) {
            // 重新检查状态
            int recheck = ctl.get();
          // 如果发现当前线程池不是处于 Running 状态，就移除之前的任务
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
             // 1. 当前线程池处于 Running 状态，但是工作线程数量为 0，需要创建新的线程
            // 2. 移除任务失败，但是工作线程数量为 0，需要创建新的线程来完成移除失败的任务
                addWorker(null, false);
        }
          //如果workQueue满了，那么这时候可能还没到线程池的maxnum，所以尝试增加一个Worker
          //线程池不是```RUNNING```状态
        else if (!addWorker(command, false))
             // 如果Worker数量到达上限，那么就拒绝此线程
            reject(command);
    }

private boolean addWorker(Runnable firstTask, boolean core) {
    // 这里一大段的 for 语句, 其实就是判断和处理 core 参数的.
    // 当经过判断, 如果当前的线程大于 corePoolSize 或 maximumPoolSize 时(根据 core 的值来判断), 
    // 则表示不能新建新的 Worker 线程, 此时返回 false.
   // retry 类似于 goto，continue retry 跳转到 retry 定义， 而 break retry 跳出 retry
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            // 当 core 为真, 那么就判断当前线程是否大于 corePoolSize
            // 当 core 为假, 那么就判断当前线程数是否大于 maximumPoolSize
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : ma))
                return false;
            // 使用 CAS 方式将线程数量增加，如果成功就跳出 retry，如果失败，证明有竞争，那么重新到retry。
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();  // Re-read ctl
             // 如果线程池运行状态发生了改变就从 retry（外层循环）处重新开始，
            if (runStateOf(c) != rs)
                continue retry;

            // 程序执行到这里说 CAS 没有成功，那么就再次执行 CAS
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            // 程序用一个 HashSet 存储线程，而 HashSet 不是线程的安全的， 所以将线程加入 HashSet 的过程需要加锁。
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());

                // 1. rs < SHUTDOWN 说明程序在运行状态
                // 2. rs == SHUTDOWN 说明当前线程处于平缓关闭状态，而 firstTask == null
                // 说明当前创建的线程是为了处理任务队列中剩余的任务（故意传入 null）
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    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;
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 这里将 Worker 中的 state 设为 0，以便其他线程可以获得锁
    // 从而可以中断当前线程
    w.unlock(); // allow interrupts
    // 用来标记线程是正常退出循环还是异常退出
    boolean completedAbruptly = true;
    try {
        // 如果任务不为空，说明是刚创建线程，如果任务为空，则从队列中取任务
        // 如果队列没有任务，线程就会阻塞在这里
        while (task != null || (task = getTask()) != null) {
            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
            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;
                    // 因为 runnable 方法不能抛出 checkedException ，所以这里
                    // 将异常包装成 Error 抛出
                    throw new Error(x);
                } finally {
                    // 任务执行完之后做一些处理，默认空函数
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        //处理worker退出的逻辑
        processWorkerExit(w, completedAbruptly);
    }
}

private Runnable getTask() {
      //主要是判断后面的poll是否要超时
        boolean timedOut = false; // Did the last poll() time out?

        retry: for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // 1.rs > SHUTDOWN 所以rs至少等于STOP,这时不再处理队列中的任务
            // 2.rs = SHUTDOWN 所以rs>=STOP肯定不成立，这时还需要处理队列中的任务除非队列为空
            // 这两种情况都会返回null让runWoker退出while循环也就是当前线程结束了
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                // 递减workerCount值
                decrementWorkerCount();
                return null;
            }

            // timed 用于判断是否需要对线程进行超时控制
            boolean timed;

            // 1.RUNING状态
            // 2.SHUTDOWN状态，但队列中还有任务需要执行
            for (;;) {
                int wc = workerCountOf(c);

                 // 1. allowCoreThreadTimeOut: 为 true 说明可以对 core 线程进行超时控制
                // 2. wc > corePoolSize: 说明线程池中有非 core 线程
                timed = allowCoreThreadTimeOut || wc > corePoolSize;

                //线程数量不大于 maximumPoolSize 并且没有超时，则退出循环，否则workerCount递减，返回null,结束当前thread
                if (wc <= maximumPoolSize && !(timedOut && timed))
                    break;

                // workerCount递减，结束当前thread
                if (compareAndDecrementWorkerCount(c))
                    return null;
                c = ctl.get(); // Re-read ctl
                // 需要重新检查线程池状态，因为上述操作过程中线程池可能被SHUTDOWN
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }

            try {
                  // 如果允许超时控制，则执行 poll 方法，该方法响应超时，当 keepAliveTime 时间内
                  // 仍然没有获取到任务，就返回 null。take 方法不响应超时操作，当获取不到任务时会一直等待。
                  // 另外不管 poll 还是 take 方法都会响应中断，如果没有新的任务添加到队列中，会直接抛出 InterruptedException
                Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;// 超时
            } catch (InterruptedException retry) {
                timedOut = false;// 线程被中断重试
            }
        }
    }

   private void processWorkerExit(Worker w, boolean completedAbruptly) {
    // 如果completedAbruptly值为true，则说明线程执行时出现了异常，需要将workerCount减1；
    // 如果线程执行时没有出现异常，说明在getTask()方法中已经已经对workerCount进行了减1操作，这里就不必再减了。 
        if (completedAbruptly) 
            decrementWorkerCount();

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //统计完成的任务数
            completedTaskCount += w.completedTasks;
            // 从workers中移除，也就表示着从线程池中移除了一个工作线程
            workers.remove(w);
        } finally {
            mainLock.unlock();
        }
        // 根据线程池状态进行判断是否结束线程池
        tryTerminate();

        int c = ctl.get();
        //当线程池是RUNNING或SHUTDOWN状态时，如果worker是异常结束，直接添加一个线程；
        // 线程正常结束， 如果允许对 core 线程进行超时控制，并且任务队列中有任务， 则保证线程数量大于等于 1
        // 如果不允许对 core 进行超时控制，则保证线程数量大于等于 corePoolSize
        if (runStateLessThan(c, STOP)) {
            if (!completedAbruptly) {
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
         /*
         * 当前线程池的状态为以下几种情况时，直接返回：
         * 1. RUNNING，因为还在运行中，不能停止；
         * 2. TIDYING或TERMINATED，因为线程池中已经没有正在运行的线程了；
         * 3. SHUTDOWN并且等待队列非空，这时要执行完workQueue中的task；
         */
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;


            // 只能是以下情形会继续下面的逻辑：结束线程池。
            // 1.SHUTDOWN状态，这时不再接受新任务而且任务队列也空了
            // 2.STOP状态，当调用了shutdownNow方法

            if (workerCountOf(c) != 0) { 
              // 如果工作线程数量不为 0，中断线程池中第一个线程
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
            // 将线程状态设为 TIDYING，如果设置不成功说明线程池的状态发生了变化，需要重试
            //如果设置成功，则调用terminated方法
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                      // terminated方法默认什么都不做，留给子类实现
                        terminated();
                    } finally {
                        // 将线程状态设为 TERMINATED
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }

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;
            // 首先看当前线程是否已经中断，如果没有中断，就看线程是否处于空闲状态
            // 如果能获得线程关联的 Worker 锁，说明线程处于空闲状态，可以中断
            // 否则说明线程不能中断
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            // 如果 onlyOne 为 true，只尝试中断第一个线程
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 检查当前线程是否有关闭线程池的权限
        checkShutdownAccess();
        // 将线程池状态设为 SHUTDOWN
        advanceRunState(SHUTDOWN);
        // 中断线程，这里最终调用 interruptIdleWorkers(false);
        interruptIdleWorkers();
        // hook 方法，默认为空，让用户在线程池关闭时可以做一些操作
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

public List < Runnable > shutdownNow() {
    List < Runnable > tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 检查线程是否具有关闭线程池的权限
        checkShutdownAccess();
        // 更改线程状态
        advanceRunState(STOP);
        // 中断线程
        interruptWorkers();
        // 清除任务队列，并将任务返回
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    return tasks;
}

private void interruptWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        // 不管线程是否空闲都执行中断
        for (Worker w: workers)
            w.interruptIfStarted();
    } finally {
        mainLock.unlock();
    }
}

private List <Runnable> drainQueue() {
    BlockingQueue <Runnable> q = workQueue;
    ArrayList <Runnable> taskList = new ArrayList < Runnable > ();
    // 该方法会将阻塞队列中的所有项添加到 taskList 中
    // 然后清空任务队列，该方法是线程安全的
    q.drainTo(taskList);
    if (!q.isEmpty()) {
        // 将 List 转换为 数组，传入的 Runnable[0] 用来说明是转为 Runnable 数组
        for (Runnable r: q.toArray(new Runnable[0])) {
            if (q.remove(r))
                taskList.add(r);
        }
    }
    return taskList;
}