Java多线程 -- JUC包源码分析11 -- ThreadPoolExecutor源码分析

Posted 2020-08-11

在JUC包中，线程池部分本身有很多组件，可以说是前面所分析的各种技术的一个综合应用。从本文开始，将综合前面的知识，逐个分析线程池的各个组件。
-Executor/Executors
-ThreadPoolExecutor使用介绍
-ThreadPoolExecutor实现原理
–ThreadPoolExecutor的中断与优雅关闭 shutdown + awaitTermination
–shutdown的一个误区

---

Executor/Executors

Executor是线程池框架最基本的几个接口：

public interface Executor {
    void execute(Runnable command);
}

而Executors是线程池框架的一个工具类，利用它可以方便的创建不同策略的线程池：

//单线程线程池：corePoolSize = maxPoolSize = 1, 队列用的LinkedBlockingQueue
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new   LinkedBlockingQueue<Runnable>()));
    }

//固定数目的线程池：corePoolSize = maxPoolSize = n, 队列用的LinkedBlockingQueue
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

//1。CachedThreadPool，corePoolSize = 0,  队列为SynchronousQueue，maxPoolSize = Integer.MAX_VALUE(这也就意味着，每来一个任务，就创建一个线程。
//2。关于SynchronousQueue，后面会单独用一篇来分析。它是个特殊的队列，没本身没有容量，放进去一个，就得等有线程拿出来，才能解除阻塞
//3。从构造参数可以看出，空闲线程，60s没人用，回收
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

//单线程的，具有周期调度功能的线程池
    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }

//多线程的，具有周期调度功能的线程池
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());
    }

从上面可以看出，Executors的各个工具函数，都用的ThreadPoolExecutor/ScheduledThreadPoolExecutor这2个类，下面做详细分析。

ThreadPoolExecutor

ThreadPoolExecutor构造函数详解

下面是ThreadPoolExecutor的参数最全的构造函数，搞清楚了每个参数的含义，也就明白了线程池的各种不同策略，也就明白了上述Executors工具类中的各个工具函数。

    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;
    }

corePoolSize: 线程池始终维护的线程个数
maxPoolSize: corePooSize满了，队列也满的情况下，扩充线程至这个值
keepAliveTime/TimeUnit: maxPoolSize中的空闲线程，过多长时间销毁，总线程数收缩回corePoolSize
blockingQueue: 线程池所用的队列类型
threadFactory: 线程创建工厂，可以自定义，也有一个缺省的
RejectedExecutionHandler: corePoolSize满了，队列满了，maxPoolSize满了，最后的拒绝策略。

ThreadPool任务处理流程

从上述构造函数解释，可以看出每次submit的任务，有如下的处理流程：
step1: 判断当前线程数 >= corePoolSize。如果小于，新建线程执行；如果大于，进入step2
step2: 判断队列是否已满。未满，放入；已满，进入step3
step3: 判断当前线程数 >= maxPoolSize。如果小于，新建线程执行；如果大于，进入step4
step4: 根据拒绝策略，拒绝任务

总结一下：先判断corePoolSize, 再判断blockingQueue，再判断maxPoolSize，最后使用拒绝策略

ThreadPool的4中拒绝策略

ThreadPoolExecutor的4个内部类，分别定义了4种策略。缺省是AbortPolicy

//策略1：让调用者直接在自己的线程里面执行，线程池不做处理
    public static class CallerRunsPolicy implements RejectedExecutionHandler {

        public CallerRunsPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

//策略2：线程池直接抛异常
    public static class AbortPolicy implements RejectedExecutionHandler {

        public AbortPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }


//策略3：线程池直接把任务丢掉，当作什么也没发生
    public static class DiscardPolicy implements RejectedExecutionHandler {

        public DiscardPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

//策略4：把队列里面最老的任务删除掉，把该任务放入队列
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {

        public DiscardOldestPolicy() { }

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

ThreadPoolExecutor实现原理

一般都知道，ThreadPool的基本实现原理就是一个队列 + 一组worker线程，调用中不断往队列中放，worker线程不断去取。但在具体实现中，有不同的实现策略：
策略1： 阻塞队列 vs. 非阻塞队列
在ThreadPoolExecutor中，使用的是阻塞队列，即如下的BlockingQueue接口：

private final BlockingQueue<Runnable> workQueue;

这也就意味着，worker内部不需要自己设置wait/notify机制，它只管从队列中取，取的到执行，取不到，自动会阻塞。

也有使用非阻塞队列的，比如Tomcat 6里面的线程池实现（以后会源码详细分析），当没有请求处理时，worker内部自己实现阻塞，然后又新的请求进来，再通知woker。

策略2：新来的请求，是直接放入队列，还是先new一个新的thread？
ThreadPool的处理方式是优先new thread处理，thread count >= corePoolSize的时候，再考虑放入队列。

策略3： 无界队列 vs. 有界队列？
如果无界队列，意味着maxPoolSize的逻辑永远不会执行。这在上面的Executors中，FixedThreadPool已有所体现。

除此之外，还有诸多实现上的细节，下面代码详细分析

源码分析

//核心结构：一个BlockingQueue + 一个线程的Set +  一把锁（控制对workers, 各种threadCount的互斥访问）
public class ThreadPoolExecutor extends AbstractExecutorService {
。。。
    private final BlockingQueue<Runnable> workQueue;  

    private final ReentrantLock mainLock = new ReentrantLock();

    private final HashSet<Worker> workers = new HashSet<Worker>();
。。。
}

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {   //小于corePoolSize的判断
            if (runState == RUNNING && workQueue.offer(command)) { //入队列
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command); //进入队列之后，2次检测
            }
            else if (!addIfUnderMaximumPoolSize(command))  //小于maxPoolSize的判断
                reject(command); // 大于maxPoolSize，拒绝请求
        }
    }

    //poolSize < corePoolSize的时候，直接new Thread，加入hashSet
    private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        return t != null;
    }

    //队列满了，poolSize < maxPoolSize，再次new thread，加入hashSet
    private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (poolSize < maximumPoolSize && runState == RUNNING)
                t = addThread(firstTask);
        } finally {
            mainLock.unlock();
        }
        return t != null;
    }

Worker的实现

    private final class Worker implements Runnable {
    。。。
    private Runnable firstTask; //至所以有firstTask这个变量，是因为创建worker的时候，可以直接赋给它一个task执行；有可以不赋给task，让它自己到blockingQueue里面去循环取
    Worker(Runnable firstTask) {
            this.firstTask = firstTask;
        }

    //1个死循环，不断从blockingQueue中，取task执行。取不到，就会阻塞在getTask()里面
    public void run() {
            try {
                hasRun = true;
                Runnable task = firstTask; 
                firstTask = null;
                while (task != null || (task = getTask()) != null) {
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);  //worker线程退出
            }
        }
   。。。
   }

//getTask里面有个关键点：当poolSize <= corePoolSize时，是无限期阻塞下去，线程也就会一直存在，不会退出，死掉；当poolSize > corePoolSize或者允许coreThread也死去时，线程就只阻塞keepAliveTime的时间，时间到了，队列还是空的，没有请求，线程就退出，死掉了，同时poolSize--.

    Runnable getTask() {
        for (;;) {
            try {
                int state = runState;
                if (state > SHUTDOWN)
                    return null;
                Runnable r;
                if (state == SHUTDOWN)  
                    r = workQueue.poll();  //poll是非阻塞调用，没有直接返回null
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                    r = workQueue.poll(keepAliveTime,  TimeUnit.NANOSECONDS);  //等待1个超时时间，默认就是构造函数里面传进去的那个60s
                else
                    r = workQueue.take();  //take是阻塞调用，没有，一直阻塞
                if (r != null)
                    return r;
                if (workerCanExit()) {
                    if (runState >= SHUTDOWN) 
                        interruptIdleWorkers();
                    return null;
                }
            } catch (InterruptedException ie) {
                // On interruption, re-check runState
            }
        }
    }

中断与优雅关闭

线程池状态切换图

    volatile int runState;
    static final int RUNNING    = 0;
    static final int SHUTDOWN   = 1;
    static final int STOP       = 2;
    static final int TERMINATED = 3;

初始处于RUNNING状态，当调用shutdown()之后，切换到SHUTDOWN状态；调用shutdownNow()，切换到STOP状态。

那shutdown与shutdownNow有什么区别吗？
shutdown(): 不会清空队列里面的任务，会等所有任务执行完毕。并且它只会中断那些 > corePoolSize的idle线程

shutdownNow(): 清空队列里面所有任务，同时向所有线程发送中断信号

当队列为空 && pool也为空时，线程池进入Terminated状态。

shutdown/shutdownNow源码解析

 public void shutdown() {
    SecurityManager security = System.getSecurityManager();
    if (security != null)
            security.checkPermission(shutdownPerm);  //权限检查，check当前调用者，是否有权限关闭线程池。没有权限，抛出异常。

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (security != null) { 
                for (Worker w : workers)
                    security.checkAccess(w.thread);  //权限检查
            }

            int state = runState;
            if (state < SHUTDOWN)
                runState = SHUTDOWN;   //从running切换到shutdown。不能从stop或者terminated切换到shutdown

            try {
                for (Worker w : workers) {
                    w.interruptIfIdle();  //遍历所有线程，向其发送信号
                }
            } catch (SecurityException se) { 
                runState = state;
                throw se;
            }

            tryTerminate(); //试图终止线程池
        } finally {
            mainLock.unlock();
        }
    }

    public List<Runnable> shutdownNow() {
    SecurityManager security = System.getSecurityManager();
    if (security != null)
            security.checkPermission(shutdownPerm);

        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            if (security != null) {
                for (Worker w : workers)
                    security.checkAccess(w.thread);
            }

            int state = runState;
            if (state < STOP)
                runState = STOP;    //切换到stop状态

            try {
                for (Worker w : workers) {
                    w.interruptNow();  //变量所有线程，发中断信号，不管是否正在执行任务
                }
            } catch (SecurityException se) { // Try to back out
                runState = state;
                // tryTerminate() here would be a no-op
                throw se;
            }

            List<Runnable> tasks = drainQueue();  //清空队列请求
            tryTerminate(); // 试图终止线程池 
            return tasks;
        } finally {
            mainLock.unlock();
        }
    }

从上面，可以看出，shutdown和shutdownNow的区别有3点：
(1)一个是切换到shutdown状态，一个是切换到stop状态
(2)遍历所有线程，一个调用的interruptIfIdle, 一个调用的interruptNow。
(3)shutdownNow会清空队列中的任务

那interruptIfIdle和interruptNow有什么区别呢？

    private final class Worker implements Runnable {

        。。。
        private final ReentrantLock runLock = new ReentrantLock();

        void interruptIfIdle() {
            final ReentrantLock runLock = this.runLock;
            if (runLock.tryLock()) {
                try {
                    if (hasRun && thread != Thread.currentThread())
                        thread.interrupt();
                } finally {
                    runLock.unlock();
                }
            }
        }

        void interruptNow() {
            if (hasRun)
                thread.interrupt();
        }

        public void run() {
            try {
                hasRun = true;
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) { //getTask内部，也有响应中断的逻辑
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);
            }
        }

        //每次从队列中拿出一个任务，执行之前，会加锁
         private void runTask(Runnable task) {
            final ReentrantLock runLock = this.runLock;
            runLock.lock();
            try {
                if ((runState >= STOP ||
                    (Thread.interrupted() && runState >= STOP)) &&
                    hasRun)
                    thread.interrupt();

                boolean ran = false;
                beforeExecute(thread, task);
                try {
                    task.run();
                    ran = true;
                    afterExecute(task, null);
                    ++completedTasks;
                } catch (RuntimeException ex) {
                    if (!ran)
                        afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                runLock.unlock();
            }
        }

可以看出，interruptIfIdle和interuptNow的关键区别是：前者会加锁访问，这也就意味着，如果被中断的线程，正在执行runTask，则锁是拿不到的。此时shutdown会阻塞，直到woker执行完runTask。

shutdown的一个误区

根据上面分析，是不是shutdown一定会阻塞到队列中所有请求都执行完，再返回呢？或者说，shutdown返回的时候，是不是队列里面的请求就一定执行完了呢？

不一定！shutdown返回之后，线程池不一定立即关闭！为什么呢？

请看下面的getTask函数

    Runnable getTask() {
        for (;;) {
            try {
                int state = runState;
                if (state > SHUTDOWN)
                    return null;
                Runnable r;
                if (state == SHUTDOWN)  //如果线程池是shutdown状态，就不阻塞了，不管是否能拿到，都是直接返回
                    r = workQueue.poll();  //关键点：如果是shutdown状态，会一直循环，直到拿空队列里面所有任务
                else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
                    r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
                else
                    r = workQueue.take();  //case1: 别的线程先置了中断标志位，然后当前线程调用take  //case 2:  先调用take阻塞在这，然后别的线程置了中断标志位  //2种case，都会抛出异常，进入下面的InterruptedException
                if (r != null)
                    return r;
                if (workerCanExit()) {
                    if (runState >= SHUTDOWN) // Wake up others
                        interruptIdleWorkers();
                    return null;
                }
                // Else retry
            } catch (InterruptedException ie) {   //阻塞的时候，收到中断，不处理，再次循环检查
                // On interruption, re-check runState
            }
        }
    }

        public void run() {
            try {
                hasRun = true;
                Runnable task = firstTask;
                firstTask = null;
                while (task != null || (task = getTask()) != null) { //getTask内部，也有响应中断的逻辑
                    runTask(task);
                    task = null;
                }
            } finally {
                workerDone(this);
            }
        }

        //每次从队列中拿出一个任务，执行之前，会加锁
         private void runTask(Runnable task) {
            final ReentrantLock runLock = this.runLock;
            runLock.lock();
            try {
                if ((runState >= STOP ||
                    (Thread.interrupted() && runState >= STOP)) &&
                    hasRun)
                    thread.interrupt();

                boolean ran = false;
                beforeExecute(thread, task);
                try {
                    task.run();
                    ran = true;
                    afterExecute(task, null);
                    ++completedTasks;
                } catch (RuntimeException ex) {
                    if (!ran)
                        afterExecute(task, ex);
                    throw ex;
                }
            } finally {
                runLock.unlock();
            }
        }

总结一下：当执行shutdown的时候，woker线程可能处于以下几种情况：
情况1： 正在执行runTask，此时拿着runLock锁，调用者会阻塞在shutdown上面。
情况2： 正要进入getTask。runTask执行完了，锁释放了，正要去getTask。此时shutdown不会阻塞，状态切换到shutdown状态，就返回了。 而getTask里面，会调用blockingQueue.poll
情况3： 在getTask里面，阻塞在blockQueue.take上面。此时调用shutdown, getTask里面收到中断，再次开始for(;;)循环

情况2，情况3，shutdown不会阻塞，就返回了。

所以不管是shutdown, 还是shutdownNow()，结尾都调用了tryTeminate，下面看看这个函数：

    private void tryTerminate() {
        if (poolSize == 0) {  //线程池里线程没了
            int state = runState;
            if (state < STOP && !workQueue.isEmpty()) {
                state = RUNNING; //关键点：线程池里线程没了，状态是shutdown状态，队列还不为空，此时把状态切会到Running状态。并且重新创建线程，消化队列中的任务
                addThread(null);
            }
            if (state == STOP || state == SHUTDOWN) {
                runState = TERMINATED;
                termination.signalAll();  //通知awaitTermination函数，不要再等了，线程池关闭
                terminated();
            }
        }
    }

所以，正确的使用shutdown的方式，应该是如下代码：

     executor.shutdown();//只是不能再提交新任务，等待执行的任务不受影响  

    //调完shutdown，要循环调用awaitTermination，等待线程池真的终止
        try {  
            boolean loop = true;  
            do {    //等待所有任务完成  
                loop = !executor.awaitTermination(2, TimeUnit.SECONDS);  //阻塞，直到线程池里所有任务结束
            } while(loop);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        } 

总结：无论是shutdown，还是shutdownDown，都无法保证线程池立即关闭。他们的本质都只是切换了线程池的状态，发送了中断信号，然后等队列里面的任务为空了，所有线程自己销毁自己。

要让主线程等待线程池彻底终止，需要调用awaitTermination函数。

关于SechduledThreadPoolExecutor，会在接下来的篇章中，详细的单独阐述。