Java中的线程池——ThreadPoolExecutor的原理

Posted 2021-01-14

1 线程池的处理流程
向线程池提交一个任务后，它的主要处理流程如下图所示

一个线程从被提交（submit）到执行共经历以下流程：

线程池判断核心线程池里是的线程是否都在执行任务，如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下一个流程
线程池判断工作队列是否已满。如果工作队列没有满，则将新提交的任务储存在这个工作队列里。如果工作队列满了，则进入下一个流程。
线程池判断其内部线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已满了，则交给饱和策略来处理这个任务。
线程池在执行excute方法时，主要有以下四种情况

1 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需要获得全局锁）
2 如果运行的线程等于或多于corePoolSize ,则将任务加入BlockingQueue
3 如果无法将任务加入BlockingQueue(队列已满)，则创建新的线程来处理任务（需要获得全局锁）
4 如果创建新线程将使当前运行的线程超出maxiumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

线程池采取上述的流程进行设计是为了减少获取全局锁的次数。在线程池完成预热（当前运行的线程数大于或等于corePoolSize）之后，几乎所有的excute方法调用都执行步骤2。

2 线程池的源码分析
2.1 定义的几个变量

 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits
    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;

    // Packing and unpacking ctl
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    private static int ctlOf(int rs, int wc) { return rs | wc; }

在分析源码前有必要理解一个变量ctl。这是Java大神们为了把工作线程数量和线程池状态放在一个int类型变量里存储而设置的一个原子类型的变量。 在ctl中，低位的29位表示工作线程的数量，高位用来表示RUNNING、SHUTDOWN、STOP等状态。 因此一个线程池的数量也就变成了(2^29)-1，大约500 million，而不是(2^31)-1，2billion。上面定义的三个方法只是为了计算得到线程池的状态和工作线程的数量。

2.2 Execute 方法提交任务

2.2.1 Execute方法

public void execute(Runnable command) {
     //如果提交了空的任务 抛出异常
        if (command == null)
            throw new NullPointerException();
     int c = ctl.get();//获取当前线程池的状态
     //检查当前工作线程数量是否小于核心线程数量
        if (workerCountOf(c) < corePoolSize) {
        //通过addWorker方法提交任务
            if (addWorker(command, true))
                return;
            c = ctl.get();//如果提交失败 需要二次检查状态
        }
    //向工作线程提交任务 
        if (isRunning(c) && workQueue.offer(command)) {
            // 再次检查状态
        int recheck = ctl.get();

            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
    //扩容失败 则拒绝任务
        else if (!addWorker(command, false))
            reject(command);
    }

从源码中可以看到提交任务的这一过程基本与第二个图的四个流程是一致的，需要检查的是当前工作线程的数量与核心线程数量的关系，来决定提交任务的方式或者是拒绝任务提交。而具体任务的提交工作是在addWorker方法中。在这里面看到了recheck这样的变量，这是在执行了一些动作失败后再次检查线程池的状态，因为在这期间可能有线程池关闭获得线程池饱和等状态的改变。

2.2.2 addWorker方法

这个方法是任务提交的一个核心方法。在里面完成了状态检查、新建任务、执行任务等一系列动作。它有两个参数，第一个参数是提交的任务，第二个参数是一个标识符，标识在检查工作线程数量的时候是应该与corePoolSize对比还是应该maximumPoolSize对比。

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
//死循环更新状态
            for (;;) {
                    int c = ctl.get();
                    int rs = runStateOf(c);//获取运行状态

        //检查线程池是否处于关闭状态
                    if (rs >= SHUTDOWN &&
                            ! (rs == SHUTDOWN &&
                                 firstTask == null &&
                                 ! workQueue.isEmpty()))
                            return false;

                    for (;;) {
        //获取当前工作线程数量
                            int wc = workerCountOf(c);
    //如果已经超过corePoolSize获取maximumPoolSize 返回false
                            if (wc >= CAPACITY ||
                                    wc >= (core ? corePoolSize : maximumPoolSize))
                                    return false;
    //CAS增加一个工作线程
                            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
                    }
            }

            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
                    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();
        //添加工作这到hashset中保存
                                            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;
    }

这个方法可以分为两个阶段来看，第一个阶段是判断是否有必要新增一个工作线程，如果有则利用CAS更新工作线程的数量；第二部分是将提交的任务封装成一个工作线程Worker然后加入到线程池的容器中，开始执行新提交的任务。这个Worker在执行完任务后，还会循环地获取工作队列里的任务来执行。下面来看一下Worker的构造方法就能更好地理解上面的代码了
/**

• Creates with given first task and thread from ThreadFactory.
• @param firstTask the first task (null if none)
  */
  Worker(Runnable firstTask) {
  setState(-1); // inhibit interrupts until runWorker
  this.firstTask = firstTask;
  this.thread = getThreadFactory().newThread(this);
  }

2.2.3?runWorker方法

在addWorker方法快要结束的地方，调用了t.start()方法，我们知道它实际执行的就是Worker对象的run()方法，而worker的run()方法是这样定义的：
/* Delegates main run loop to outer runWorker /
public void run() {
runWorker(this);
}
它实际上是将自己委托给线程池的runWorker方法

final void runWorker(Worker w) {

    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
    //不断地从blockingQueue获取任务
        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方法
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
        //调用Runable的run方法
                    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 {
        // 执行aferExecute方法
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

这个方法呢也比较好理解，它在不断执行我们提交的任务的run方法。而这个任务可能是我们新提交的，也有可能是从等待队列中获取的。这样就实现了线程池的完成逻辑。