ThreadPoolExecutor线程池运行机制分析-线程复用原理

红太狼 2021-04-27 15:06 544阅读 0赞

# 介绍 #

我们在涉及到多线程开发时，必然就涉及到线程池的使用，在阿里巴巴-Java开发手册中说到：

> 【强制】线程资源必须通过线程池提供，不允许在应用中自行显示创建线程 说明：使用线程池的好处是减少在创建和销毁线程上所花费的时间以及系统资源的开销，解决资源不足的问题，如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或“过度切换”问题。

这里的规范很强势，直接定义为【强制】，目的就是使用线程复用机制减少线程开销。

关于线程池的线程复用机制就是我们这篇博文讨论的目的。

# 分析代码 #

我们在分析代码写博文时，要抓住关键点，而不是面面俱到，这样才是思路清晰的代码分析。

我这里尽量简洁，从代码入手，但不是一行行一个个解释知识点，所以尽量以文字说明，搭配少量代码力求简洁。

# 入口 #

从ThreadPoolExecutor\#execute(Runnable)方法为入口 要了解线程池的线程复用机制，首先要了解线程池中线程是如何创建，简单说就是两种：

1.  线程池中的线程数量小于core线程数限定，直接addWorker创建新线程，并返回。
2.  留待后文
3.  线程池中的线程数量小于max线程数限定，还能够继续创建新线程。否则抛出拒绝异常。

以上的1、3情况都比较简单，没有涉及到线程复用和任务队列

对应到代码

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */
            int c = ctl.get();
            //1：创建core核心线程
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            //2：情况复杂后文分析
            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);
            }
            //创建work工作线程
            else if (!addWorker(command, false))
                reject(command);
        }

# 任务队列 #

上文的的2情况比较复杂：线程池中线程数量已经超过core核心数限定，线程池还在运行中（注：不是线程），且Runnable任务能够成功入队任务队列`workQueue`。

## 关于入队 ##

关于成功入队有两种类型的队列：

*  普通的阻塞队列：如ArrayBlockingQueue，LinkedBlockingQueue，PriorityBlockingQueue等，只要队列的容量足够就能成功入队。
 *  其他阻塞队列：就是SynchronousQueue，它的成功入队表示有线程同时在接收入队的数据，有线程能处理入队数据。这里留下后文解释，这个是解线程复用的关键。

总之只要能够任务能成功入队就表示线程池还有能力处理新任务，但是还是需要注意：

> 如果是队列有容量能入队，但是实际上线程池中并没有线程在运行，就需要`addWorker`一个null空的`firstTask`的线程去取队列中的任务执行。这一步其实还是创建了新线程。但是它的思路非常接近线程复用。

# 线程的开启和创建 #

现在就剩一个问题了：任务入队，且有线程池中有线程在运行这是什么情况？ 这个情况就是复用线程的情况。

这里就需要深入到源码`addWorker()`：它是创建新线程的关键，也是线程复用的关键入口。

*  创建线程：new创建Woker调用的初始化方法，就是创建新线程过程。
 *  线程运行：注意这里创建Woker时只是指定了firstTask（可为null）第一个任务，然后start运行线程

对应代码： 线程开启运行过程：

private boolean addWorker(Runnable firstTask, boolean core) {
            //检查和cas同步操作 省略
            boolean workerStarted = false;
            Worker w = null;
            try {
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
                    //添加和扩容操作 省略
                    if (workerAdded) {
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                //处理失败操作
            }
            //返回新线程是否成功开启
            return workerStarted;
        }

创建过程：

/** * 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);
            }

# 线程运行 #

1.  从线程开始start()
2.  Worker implements Runnable,Runable的执行run()
3.  最后来到runWoker(Worker)

runWoker取任务有两个方式：

*  firstTask：这是指定的第一个runnable可执行任务，它会在Woker这个工作线程中运行执行任务run。并且置空表示这个任务已经被执行。
 *  getTask()：这首先是一个死循环过程，工作线程循环直到能够取出Runnable对象或超时返回，这里的取的目标就是任务队列workQueue，对应刚才入队的操作，有入有出。

对应代码：

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            //省略无关代码
             while (task != null || (task = getTask()) != null) {
                 try {
                      task.run();
                    } finally {
                        task = null;
                    }
        }

# 线程存活 #

这个点太过关键，所以先贴代码感受一下。

private Runnable getTask() {
            boolean timedOut = false; // Did the last poll() time out?
    
            for (;;) {
                int c = ctl.get();
               //检查操作 省略 
               //是否开启 时间检测机制
                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;
                }
            }
        }

最后`getTask`()\`就是整个问题关键：取任务的方式

*  无限定时间：默认不允许core核心线程超时allowCoreThreadTimeOut=false（允许的话就是所有的线程都有超时机制），以及判定当前是core核心线程，没有限定时间的方式，就会一直在`take()`中阻塞直到取出任务。这就是核心线程默认的情况，一直阻塞一直存活一直工作。
 *  有限定时间：那就是开启allowCoreThreadTimeOut的核心线程，以及work工作线程，这是keepAliveTime有效作用的唯一地方，它决定了任务线程从任务队列取任务（出队）时`poll()`，最大的等待时间，在这段时间keepAliveTime就是线程的有限的存活时间。如果在限定时间之后取不出任务，就出现超时，对外返回null，最终让当前线程结束。

在线程长时间存活过程中，如果能从任务队列中取出任务，那么就getTask返回让任务线程执行。否则如果有机会收到返回null任务，外部的runWoker的while循环也终止，完结当前任务线程。

# 总结 #

线程池的线程复用：就是任务在并不只执行创建时指定的firstTask第一任务，还会从任务队列的中自己主动取任务执行，而且是有/无时间限定的阻塞等待，保证线程的存活。