Javaのスレッドプールのグラフィックコードの詳細説明

package org.my.threadPoolDemo;
/**
 * 任务类，实现Runnable接口 重写run方法
 */
public class MyTask implements Runnable{
    
    private int taskNum;
    
    public MyTask(int taskNum) {
        super();
        this.taskNum = taskNum;
    }

    @Override
    public void run() {
        System.out.println("正在执行task"+taskNum);
        try {
            Thread.currentThread().sleep(4000);//sleep 4秒模拟执行代码过程
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task"+taskNum+"执行完毕");
    }
}

package org.my.threadPoolDemo;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolUseDemo {

    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<Runnable>());
        for (int i = 0; i < 15; i++) {
            MyTask myTask = new MyTask(i);
            executor.execute(myTask);
            System.out.println("线程池中线程数量："+executor.getPoolSize()+",线程池中等待执行的任务数量："+executor.getQueue().size()+",已执行完的任务数量："+executor.getCompletedTaskCount());
        }
        executor.shutdown();
    }

}

正在执行task0
线程池中线程数量：1,线程池中等待执行的任务数量：0,已执行完的任务数量：0
线程池中线程数量：2,线程池中等待执行的任务数量：0,已执行完的任务数量：0
正在执行task1
线程池中线程数量：3,线程池中等待执行的任务数量：0,已执行完的任务数量：0
正在执行task2
线程池中线程数量：4,线程池中等待执行的任务数量：0,已执行完的任务数量：0
正在执行task3
线程池中线程数量：5,线程池中等待执行的任务数量：0,已执行完的任务数量：0
正在执行task4
线程池中线程数量：5,线程池中等待执行的任务数量：1,已执行完的任务数量：0
线程池中线程数量：5,线程池中等待执行的任务数量：2,已执行完的任务数量：0
线程池中线程数量：5,线程池中等待执行的任务数量：3,已执行完的任务数量：0
线程池中线程数量：5,线程池中等待执行的任务数量：4,已执行完的任务数量：0
线程池中线程数量：5,线程池中等待执行的任务数量：5,已执行完的任务数量：0
线程池中线程数量：6,线程池中等待执行的任务数量：5,已执行完的任务数量：0
正在执行task10
线程池中线程数量：7,线程池中等待执行的任务数量：5,已执行完的任务数量：0
正在执行task11
线程池中线程数量：8,线程池中等待执行的任务数量：5,已执行完的任务数量：0
正在执行task12
正在执行task13
线程池中线程数量：9,线程池中等待执行的任务数量：5,已执行完的任务数量：0
线程池中线程数量：10,线程池中等待执行的任务数量：5,已执行完的任务数量：0
正在执行task14
task1执行完毕
task10执行完毕
正在执行task5
task11执行完毕
task13执行完毕
task4执行完毕
task3执行完毕
task2执行完毕
task0执行完毕
正在执行task9
正在执行task8
正在执行task7
task14执行完毕
task12执行完毕
正在执行task6
task5执行完毕
task9执行完毕
task7执行完毕
task8执行完毕
task6执行完毕

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);
    }
    
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,threadFactory, defaultHandler);
    }
    
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), handler);
    }
    
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.acc = System.getSecurityManager() == null ?null :AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

//Integer.SIZE是一个静态常量，值为32，也就是说COUNT_BITS是29
private static final int COUNT_BITS = Integer.SIZE - 3;
//CAPACITY是最大容量536870911，因为1左移29位之后-1，导致最高三位为0，而其余29位全部为1
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
//ctl用于表示线程池状态和有效线程数量，最高三位表示线程池的状态，其余低位表示有效线程数量
//初始化之后ctl等于RUNNING的值，即默认状态是运行状态，线程数量为0
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//1110 0000 0000 0000 0000 0000 0000 0000最高三位为111
private static final int RUNNING    = -1 << COUNT_BITS;
//最高三位为000
private static final int SHUTDOWN   =  0 << COUNT_BITS;
//0010 0000 0000 0000 0000 0000 0000 0000最高三位为001
private static final int STOP       =  1 << COUNT_BITS;
//0100 0000 0000 0000 0000 0000 0000 0000最高三位为010
private static final int TIDYING    =  2 << COUNT_BITS;
//0110 0000 0000 0000 0000 0000 0000 0000最高三位为011
private static final int TERMINATED =  3 << COUNT_BITS;
/**
* 获取运行状态，入参为ctl。因为CAPACITY是最高三位为0，其余低位为1
* 所以当取反的时候，就只有最高三位为1，再经过与运算，只会取到ctl的最高三位
* 而这最高三位如上文所述，表示线程池的状态
*/
private static int runStateOf(int c)     { return c & ~CAPACITY; }
/**
* 获取工作线程数量，入参为ctl。因为CAPACITY是最高三位为0，其余低位为1
* 所以当进行与运算的时候，只会取到低29位，这29位正好表示有效线程数量
*/
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
//任务队列，用于存放待执行任务的阻塞队列
private final BlockingQueue<Runnable> workQueue;
/** 判断线程池是否是运行状态，传入的参数是ctl的值
* 只有RUNNING的符号位是1，也就是只有RUNNING为负数
* 所以如果目前的ctl值<0，就是RUNNING状态
*/
private static boolean isRunning(int c) {
    return c < SHUTDOWN;
}
//从任务队列中移除任务
public boolean remove(Runnable task) {
    boolean removed = workQueue.remove(task);
    tryTerminate(); // In case SHUTDOWN and now empty
    return removed;
}

private Runnable getTask() {
        //记录循环体中上个循环在从阻塞队列中取任务时是否超时
        boolean timedOut = false; 
        //无条件循环，主要是在线程池运行正常情况下
        //通过循环体内部的阻塞队列的阻塞时间，来控制当前线程的超时时间
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);//获取线程池状态

            /*先获取线程池的状态，如果状态大于等于STOP，也就是STOP、TIDYING、TERMINATED之一
             *这时候不管队列中有没有任务，都不用去执行了；
             *如果线程池的状态为SHUTDOWN且队列中没有任务了，也不用继续执行了
             *将工作线程数量-1，并且返回null
             **/
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            //获取工作线程数量
            int wc = workerCountOf(c);

            //是否启用超时参数keepAliveTime
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            //如果这个条件成立，如果工作线程数量-1成功，返回null，否则跳出循环
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            //如果需要采用阻塞形式获取，那么就poll设定阻塞时间，否则take无限期等待。
            //这里通过阻塞时间，间接控制了超时的值，如果取值超时，意味着这个线程在超时时间内处于空闲状态
            //那么下一个循环，将会return null并且线程数量-1
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;//获取传入的Worker对象中的任务
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try {
                /* 
                 * 如果传入的任务为null，就从任务队列中获取任务，只要这两者有一个不为null，就进入循环体
                 */
            while (task != null || (task = getTask()) != null) {
                w.lock();
                //如果线程池状态已被标为停止，那么则不允许该线程继续执行任务！或者该线程已是中断状态,
                //也不允许执行任务,还需要中断该线程!
                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++;
                    w.unlock();
                }
                /*
                 * 从这里可以看出，当一个任务执行完毕之后，循环并没有退出，此时会再次执行条件判断
                 * 也就是说如果执行完第一个任务之后，任务队列中还有任务，那么将会继续在这个线程执行。
                 * 这里也是很巧妙的地方，不需要额外开一个控制线程来看那些线程处于空闲状态，然后给他分配任务。
                 * 直接自己去任务队列拿任务
                 */
            }
            completedAbruptly = false;
        } finally {
                /*
                 * 执行到这里，说明getTask返回了null，要么是超时（任务队列没有任务了），要么是线程池状态有问题了
                 * 当前线程将被回收了
                 */
            processWorkerExit(w, completedAbruptly);
        }
    }

private boolean addWorker(Runnable firstTask, boolean core) {
        //上来就是retry，后面continue retry;语句执行之后都会从这里重新开始
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);//获取线程池运行状态

            /*
             * 获取当前线程池的状态，如果是STOP，TIDYING,TERMINATED状态的话，则会返回false
             * 如果现在状态是SHUTDOWN，但是firstTask不为空或者workQueue为空的话，那么直接返回false
             */
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);//获取工作线程数量
                /*
                 * addWorker传入的第二个Boolean参数用来判别当前线程数量是否大于核心池数量
                 * true，代表当前线程数量小于核心池数量
                 */
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //增加工作线程数量
                if (compareAndIncrementWorkerCount(c))
                        //如果成功，跳出retry
                    break retry;
                c = ctl.get();  // Re-read ctl
                //判断线程池状态，改变了就重试一次
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
                /*
                 * 前面顺利增加了工作线程数，那么这里就真正创建Worker
                 * 上面流程图中说过，创建Worker时会创建新的线程.如果这里创建失败
                 * finally中会将工作线程数-1
                 */
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                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());

                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//将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;
    }

 public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        //获取到成员变量ctl的值（线程池状态）
        int c = ctl.get();
        //如果工作线程的数量<核心池的大小
        if (workerCountOf(c) < corePoolSize) {
            //调用addWorker（这里传入的true代表工作线程数量<核心池大小）
            //如果成功，方法结束。
            if (addWorker(command, true))
                return;
            //否则，再重新获取一次ctl的值
            //个人理解是防止前面这段代码执行的时候有其他线程改变了ctl的值。
            c = ctl.get();
        }
        //如果工作线程数量>=核心池的大小或者上一步调用addWorker返回false，继续走到下面
        //如果线程池处于运行状态，并且成功将当前任务放入任务队列
        if (isRunning(c) && workQueue.offer(command)) {
            //为了线程安全，重新获取ctl的值
            int recheck = ctl.get();
            //如果线程池不处于运行状态并且任务从任务队列移除成功
            if (! isRunning(recheck) && remove(command))
                //调用reject拒绝执行，根据handler的实现类抛出异常或者其他操作
                reject(command);
            //否则，如果工作线程数量==0，调用addWorker并传入null和false
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //执行到这里代表当前线程已超越了核心线程且任务提交到任务队列失败。（可以注意这里的addWorker是false）
        //那么这里再次调用addWroker创建新线程（这时创建的线程是maximumPoolSize）。
        //如果还是提交任务失败则调用reject处理失败任务
        else if (!addWorker(command, false))
            reject(command);
    }