Java 线程池 ThreadPoolExecutor源码简析

Posted 郭梧悠

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java 线程池 ThreadPoolExecutor源码简析相关的知识,希望对你有一定的参考价值。

如果你对线程池的概念还不甚了解,那么Java 线程池 Executor浅入浅出这篇文章可以让你一步步从0到1了解线程池的搭建过程。关于线程池的更详细的说明,《Java并发编程实战》这本书上讲的很透彻,强烈推荐阅读。

Executors中为我们提供了多种静态工厂方法来创建各种特性的线程池,其中大多数是返回ThreadPoolExecutor对象。因此本篇博文直接对ThreadPoolExecutor的原理进行剖析,加深对线程池设计的理解。ThreadPoolExecutor有如下两个重要参数:

corePoolSize

线程池核心线程的数量。如果线程池里得线程个数小于corePoolSize,那么就会立即创建一个新线程对象来执行任务。如果任务数大于核心线程数怎么办呢?那么就将任务放在队列中,等待执行。
比如我们去银行办理业务,有若干个人工服务窗口,当来办理业务的客户比较少的时候,可以不用排队,直接去空闲的窗口去办理业务。如果客户比较多了,那么就需要就需要领号排队了。但是服务窗口有限,来办理业务的人太多了,怎么办?maximumPoolSize就派上用场了。
我们需要注意的是,初始化ThreadPoolExecutor的时候,线程Thread不会立即创建和启动,而是等到提交任务的时候才会创建启动,当然调用了prestartAllCoreThreads的话,是可以创建并启动所有Thread.

    /**
     *启动所有核心线程,让他们处于idly状态等待任务的到来。
     * @return 返回启动的线程的数量
     */
    public int prestartAllCoreThreads() {
        int n = 0;
        while (addWorker(null, true))
            ++n;
        return n;
    }

另外ThreadPoolExecutor提供了getPoolSize()方法,该方法并不是返回初始化的时候设置的corePoolSize的大小,而是获取线程中实际存在的线程的数量,如果线程池处于TIDYING状态,则返回0:

   //存储正在工作的线程
    private final HashSet<Worker> workers = new HashSet<>();

    /**
     * Returns the current number of threads in the pool.
     *
     * @return the number of threads
     */
    public int getPoolSize() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //如果线程处于TIDYING状态,说明线程池里的线程为0
            return runStateAtLeast(ctl.get(), TIDYING) ? 0
                : workers.size();
        } finally {
            mainLock.unlock();
        }
    }

maximumPoolSize

既然有核心线程,那么也有非核心线程,但是这个线程池这个池子不能无限大,maximumPoolSize就限定了线程池中允许的最大线程数。
在线程池阻塞队列(非无界队列)装满了任务之后,如果还有任务加入,如果当前的线程数小于maximumPoolSize,则会新建线程来执行任务。否则就会拒绝执行。
比如去银行排队办理业务的客户越来越多,银行服务大厅已经装不下了。此时如果除了服务窗口之外,银行还有额外的资源可以办理客户的业务,那么就使用额外的资源来处理任务。如果额外的资源也已经满负荷服务客户,对于后面新来的客户,只能“欢迎下次光临”了。

workQueue

用来保存等待执行任务的队列,等待的任务必须实现Runnable接口,线程池与工作队列WorkQueue密切相关,因为并不是一个线程负责处理一个任务,而是一个线程负责处理好多任务。WorkQueue负责保存所有等待执行的任务,而线程池里的线程从工作队列中获取一个任务,然后执行,执行完毕后等待下一个任务。我们先来看看工作线程(下文简称WorkThread)怎么执行任务的,具体的在ThreadPoolExecutorl类的runWorker方法里面,该方法的主要作用就是不断的从WorkQueue中获取任务并执行之

  final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 释放锁,允许中断
        boolean completedAbruptly = true;
        try {
           //开启一个循环,从WorkQueue中获取任务并执行之
            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;
                    //执行任务
                     task.run();
         
                     //空方法:任务执行完毕之后的回调
                        afterExecute(task, thrown);
                    }
                } finally {
                    task = null;
                    //记录当前Worker已经完成任务的数量
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
           //使得工作线程退出
            processWorkerExit(w, completedAbruptly);
        }
    }

注意Worker 是一个Runnable,runWorker方法就是在Worker的run方法中执行的。

Worker对象简单说明

Worker是线程池里一个重要的对象,该类还是一个Runnable,该类持有一个Thread,代表着Worker对象在哪个线程中执行,在初始化Worker对象的时候就创建了线程。.

{
        //由ThreadFactory创建,代表着Worker对象在哪个线程中执行
        final Thread thread;
        /** Initial task to run.  Possibly null.
        Worker对象第一个执行的任务,可能是null,后面在执行任务的时候就要从WorkerQueue里面获取 */
        Runnable firstTask;
        /** 记录每个线程完成工作的数量*/
        volatile long completedTasks;

     
        Worker(Runnable firstTask) {
            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);
        }
        // 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) {
                }
            }
        }
    }

另外Worker提供了CAS算法进行自旋,提供了lockunlock等方法,其中1代表着处于locked状态,而0除以非locked状态.关于自旋CAS的基础知识,感兴趣的可以阅读博主的《java线程知识点拾遗(CAS)》。

execute方法

ThreadPoolExecutor提供了execute方法,主要是用新的线程或者线程池里已经有的线程,在未来某个时期执行一个任务,当线程池shutDown或者达到最大容量的时候,任务就不会被执行,此时线程池会执行拒绝策略RejectedExecutionHandler。下面再来大致分析下该方法:

 
  public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
       
        int c = ctl.get();
        //如果线程池里的线程小于核心数量
        if (workerCountOf(c) < corePoolSize) {
            //创建一个新线程,并将command作为线程执行的第一个任务
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果线程池处于Running状态,且将任务command成功加入WorkerQueue中
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //再次校验状态,如果线程池不是处于Running状态,则从队列中移除并执行拒绝策略。
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)//注意这是一个很细节的点
                //线程池没有活动线程了,但是队列里还有任务没执行这种特殊情况,则重新添加一个线程执行。
                addWorker(null, false);
        }
        //队列已经满了,则重新创建线程执行,如果失败则执行拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

在这里有一个值得注意的细节,调用workQueue.offer(command)将任务添加到队列之后,如果线程中有正在执行任务的线程,则这个任务会在未来某个时间内执行。但是如果任务入队之后,恰好线程池中没有活跃的线程了,怎么办呢,就再次addWorker,启动新线程执行。(举个不恰当的例子,就好像去即将打烊的饭店去吃饭,你已经扫码付好钱(入队列),然而厨师恰巧都已经准备下班了,不得不重新给你做饭一样),注意此时addWorker的第一个参数是null,因为我们已经将command通过offer添加到队列中了,具体代码段如下:

  else if (workerCountOf(recheck) == 0)//注意这是一个很细节的点
                //线程池没有活动线程了,但是队列里还有任务没执行这种特殊情况,则重新添加一个线程执行。
                addWorker(null, false)

可以用如下流程图来大致表示下线程池的工作流程。

addWorker方法简析

addWorker的主要作用就是,在线程池当前的状态和既有的线程边界(corePoolSize,maximumPoolSize)下,新的任务是否能添加成功。

     *
     * @param firstTask  当Worker被创建的时候(此时线程池里的线程数量小于核心线程),firstTask将作为Worker对象的第一个任务来执行。
     * 此时firstTask并不会添加到队列中。 或者当队列已经满的情况下,也是直接创建Worker,但是fistTask也不会添加到WorkQueue中去。 
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {// 外层自旋
            int c = ctl.get();
            //获取线程状态
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //如果线程池处于关闭状态,并且WorkerQueue中还有任务在等待执行,则返回false
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {//内层自旋
                int wc = workerCountOf(c);
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    //如果线程池内的线程达到最大值,比如达到corePoolSize或者maximunPoolSize,则返回false
                    return false;
                if (compareAndIncrementWorkerCount(c))
                 // 以CAS的方式尝试把线程数加1
            	 // 注意这里只是把线程池中的线程数加1,并没有在线程池中真正的创建线程
                    break retry;
                c = ctl.get();  // Re-read ctl
                   //如果此时线程池的状态和上次获取的不一样,则继续执行外层自旋
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }//end inner for
        }//end outter for

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
           //创建Worker,同时也就创建了Thread对象
            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();
                        //将Worker添加到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;
    }

addWorker工作的大致流程如下:

1、判断线程池状态,如果需要创建Worker对象并同时创建工作线程
2、将此worker添加进workers集合中。
3、启动worker对应的线程,运行worker的run,最终执行了runWorker方法。
4、如果失败了回滚worker的创建动作(addWorkerFailed 方法),即将worker从workers集合中删除,并原子性的减少workerCount。

keepAliveTime的使用原理

该变量表示线程空闲的时间。线程的创建和销毁是需要代价的。线程执行完任务后不会立即销毁,而是继续存活一段时间,存活的时间就是keepAliveTime。默认情况下,该参数只有在线程数大于corePoolSize时才会生效.该变量在从WorkQueue里获取任务的时候会被使用,在前文runWorker里就调用了从workQueue里面获取任务的方法:getTask():


private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
    for (;;) {//开启一个循环
        int c = ctl.get();
        int rs = runStateOf(c);

        //当线程处于SHUTDOWN 状态时,并且workQueue请求队列为空,线程数量的表示-1,并返回一个null。
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        //获取当前工作线程数
        int wc = workerCountOf(c);

     
        //如果allowCoreThreadTimeOut设置为true,或者线程数量大于核心线程数量,keepAliveTime才起作用
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            //减少工作线程
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
            //从阻塞队列中取任务,根据情况选择一直阻塞等待还是超时等待
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            //成功从队列中获取到了任务,并返回之    
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

可以看出,keepAliveTime是在 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)方法中起作用的。在runWorker方法中,开始while循环获取任务,如果超时了就可以跳出while循环,从而Thread完成了其任务而被销毁。:

 final void runWorker(Worker w) {
      //省略部分代码
      while (task != null || (task = getTask()) != null) {
       //省略部分代码
      }
       //省略部分代码
    }

以上是关于Java 线程池 ThreadPoolExecutor源码简析的主要内容,如果未能解决你的问题,请参考以下文章

安卓线程池ThreadPoolExecutor的常见使用

多线程(基础2)

多线程(基础2)

关于java线程池

java 如何获得线程池中正在执行的线程数?

Java线程池