首页 > 代码库 > Java多线程 -- JUC包源码分析11 -- ThreadPoolExecutor源码分析

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

技术分享

在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,会在接下来的篇章中,详细的单独阐述。

<script type="text/javascript"> $(function () { $(‘pre.prettyprint code‘).each(function () { var lines = $(this).text().split(‘\n‘).length; var $numbering = $(‘
    ‘).addClass(‘pre-numbering‘).hide(); $(this).addClass(‘has-numbering‘).parent().append($numbering); for (i = 1; i <= lines; i++) { $numbering.append($(‘
  • ‘).text(i)); }; $numbering.fadeIn(1700); }); }); </script>

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