文章目录

线程池原理
一、为什么要使用线程池
二、线程池的原理
- 1. ThreadPoolExecutor提供的构造方法
- - 1.1 构造方法
  - 1.2 构造方法参数
- 2. ThreadPoolExecutor的策略
- 3. 线程池主要的任务处理流程
- 4.ThreadPoolExecutor如何做到线程复用的？
三、四种常见的线程池
- 1. newCachedThreadPool
- 2. newFixedThreadPool
- 3. newSingleThreadExecutor
- 4. newScheduledThreadPool
总结

线程池原理

本章小宋讲解一下线程池的相关知识，用到多线程的同学都要去使用线程池，所以多了解它的原理还是很有必要的。

下文会带着大家从为什么用线程池到解析它的原理带大家深入理解。

一、为什么要使用线程池

我们为什么要去使用线程池呢？其实主要是三个原因:

创建/销毁线程需要消耗系统资源，线程池可以复用已创建的线程。
控制并发的数量，并发数量过多，可能会导致资源消耗过多，从而导致服务器崩溃，这也是为什么使用的主要原因
可以对线程做统一管理。

二、线程池的原理

Java中的线程池顶层接口是Executor接口，ThreadPoolExecutor是这个接口的实现类。下面我们带大家详细了解ThreadPoolExecutor。

1. ThreadPoolExecutor提供的构造方法

ThreadPoolExecutor一共提供了四个构造函数

1.1 构造方法

五个参数的构造函数
六个参数的构造函数(分别有两个）
七个参数的构造函数

1.2 构造方法参数

上面ThreadPoolExecutor的构造函数一共涉及到了7个参数。

我们首先来看一下必要的前5个参数的意思:

int corePoolSize：该线程池中核心线程数最大值
核心线程：线程池中有两类线程，核心线程和非核心线程。核心线程默认情况下会一直存在于线程池中，即使这个核心线程什么都不干（铁饭碗），而非核心线程如果长时间的闲置，就会被销毁（临时工）。
int maximumPoolSize：该线程池中线程总数最大值。
该值等于核心线程数量 + 非核心线程数量。
long keepAliveTime：非核心线程闲置超时时长。
非核心线程如果处于闲置状态超过该值，就会被销毁。如果设置allowCoreThreadTimeOut(true)，则会也作用于核心线程。
TimeUnit unit：keepAliveTime的单位。
TimeUnit是一个枚举类型，包括以下属性：
NANOSECONDS ：1微毫秒 = 1微秒 / 1000; MICROSECONDS ：1微秒 = 1毫秒 / 1000; MILLISECONDS ：1毫秒 = 1秒 /1000; SECONDS ：秒; MINUTES ：分; HOURS ：小时; DAYS ：天
BlockingQueue workQueue：阻塞队列，维护着等待执行的Runnable任务对象。
常用的几个阻塞队列：
LinkedBlockingQueue
链式阻塞队列，底层数据结构是链表，默认大小是Integer.MAX_VALUE，也可以指定大小。
ArrayBlockingQueue
数组阻塞队列，底层数据结构是数组，需要指定队列的大小。
SynchronousQueue
同步队列，内部容量为0，每个put操作必须等待一个take操作，反之亦然。
DelayQueue
延迟队列，该队列中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。

关于阻塞队列这一块我会在下一篇文章更加深入去讲解。

上面介绍了5个必要的参数，还有两个非必须的参数。

ThreadFactory threadFactory
创建线程的工厂，用于批量创建线程，统一在创建线程时设置一些参数，如是否守护线程、线程的优先级等。如果不指定，会新建一个默认的线程工厂。
RejectedExecutionHandler handler
拒绝处理策略，线程数量大于最大线程数就会采用拒绝处理策略，四种拒绝处理的策略为：
1.ThreadPoolExecutor.AbortPolicy：默认拒绝处理策略，丢弃任务并抛出RejectedExecutionException异常。
2.ThreadPoolExecutor.DiscardPolicy：丢弃新来的任务，但是不抛出异常。
3.ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列头部（最旧的）的任务，然后重新尝试执行程序（如果再次失败，重复此过程）。
4.ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务。

2. ThreadPoolExecutor的策略

线程池本身有一个调度线程，这个线程就是用于管理布控整个线程池里的各种任务和事务，例如创建线程、销毁线程、任务队列管理、线程队列管理等等。

故线程池也有自己的状态。ThreadPoolExecutor类中定义了一个volatile int变量runState来表示线程池的状态，分别为RUNNING、SHUTDOWN、STOP、TIDYING 、TERMINATED。

线程池创建后处于RUNNING状态。
调用shutdown()方法后处于SHUTDOWN状态，线程池不能接受新的任务，清除一些空闲worker,会等待阻塞队列的任务完成。
调用shutdownNow()方法后处于STOP状态，线程池不能接受新的任务，中断所有线程，阻塞队列中没有被执行的任务全部丢弃。此时，poolsize=0,阻塞队列的size也为0。
当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态。接着会执行terminated()函数。
```
  ThreadPoolExecutor中有一个控制状态的属性叫ctl，它是一个AtomicInteger类型的变量。
```
线程池处在TIDYING状态时，执行完terminated()方法之后，就会由 TIDYING -> TERMINATED，线程池被设置为TERMINATED状态。

3. 线程池主要的任务处理流程

处理任务的核心方法是execute，我们看看 JDK 1.8 源码中ThreadPoolExecutor是如何处理线程任务的：

public void execute(Runnable command) {if (command == null)throw new NullPointerException();   int c = ctl.get();// 1.当前线程数小于corePoolSize,则调用addWorker创建核心线程执行任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}// 2.如果不小于corePoolSize，则将任务添加到workQueue队列。if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 2.1 如果isRunning返回false(状态检查)，则remove这个任务，然后执行拒绝策略。if (! isRunning(recheck) && remove(command))reject(command);// 2.2 线程池处于running状态，但是没有线程，则创建线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 3.如果放入workQueue失败，则创建非核心线程执行任务，// 如果这时创建非核心线程失败(当前线程总数不小于maximumPoolSize时)，就会执行拒绝策略。else if (!addWorker(command, false))reject(command);
}

ctl.get()是获取线程池状态，用int类型表示。第二步中，入队前进行了一次isRunning判断，入队之后，又进行了一次isRunning判断。

为什么要二次检查线程池的状态?

在多线程的环境下，线程池的状态是时刻发生变化的。很有可能刚获取线程池状态后线程池状态就改变了。判断是否将command加入workqueue是线程池之前的状态。倘若没有二次检查，万一线程池处于非RUNNING状态（在多线程环境下很有可能发生），那么command永远不会执行。

处理流程:

线程总数量 < corePoolSize，无论线程是否空闲，都会新建一个核心线程执行任务（让核心线程数量快速达到corePoolSize，在核心线程数量 < corePoolSize时）。注意，这一步需要获得全局锁。
线程总数量 >= corePoolSize时，新来的线程任务会进入任务队列中等待，然后空闲的核心线程会依次去缓存队列中取任务来执行（体现了线程复用）。
当缓存队列满了，说明这个时候任务已经多到爆棚，需要一些“临时工”来执行这些任务了。于是会创建非核心线程去执行这个任务。注意，这一步需要获得全局锁。
缓存队列满了，且总线程数达到了maximumPoolSize，则会采取上面提到的拒绝策略进行处理。

如图:

4.ThreadPoolExecutor如何做到线程复用的？

大家都知道，一个线程在创建的时候会指定一个线程任务，当执行完这个线程任务之后，线程自动销毁。但是线程池却可以复用线程，即一个线程执行完线程任务后不销毁，继续执行另外的线程任务。

那么，线程池如何做到线程复用呢？

原来，ThreadPoolExecutor在创建线程时，会将线程封装成工作线程worker,并放入工作线程组中，然后这个worker反复从阻塞队列中拿任务去执行。下面带大家看看源码，前后有联系。

这里的addWorker方法在上面提到的execute方法里面调用过的，先看看addWorker方法的上半部分：

// ThreadPoolExecutor.addWorker方法源码上半部分
private boolean addWorker(Runnable firstTask, boolean core) {retry:for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;for (;;) {int wc = workerCountOf(c);if (wc >= CAPACITY ||// 1.如果core是ture,证明需要创建的线程为核心线程，则先判断当前线程是否大于核心线程// 如果core是false,证明需要创建的是非核心线程，则先判断当前线程数是否大于总线程数// 如果不小于，则返回falsewc >= (core ? corePoolSize : maximumPoolSize))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateOf(c) != rs)continue retry;// else CAS failed due to workerCount change; retry inner loop}}

上半部分主要是判断线程数量是否超出阈值，超过了就返回false。我们继续看下半部分:

 // ThreadPoolExecutor.addWorker方法源码下半部分boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {// 1.创建一个worker对象w = new Worker(firstTask);// 2.实例化一个Thread对象final Thread t = w.thread;if (t != null) {// 3.线程池全局锁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 startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {// 4.启动这个线程t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;
}

上面的源码中，创建了worker对象，并实例化了一个Thread对象，最后启动这个线程对象。

我们接着看看Worker类，展示部分源码：

// Worker类部分源码
private final class Worker extends AbstractQueuedSynchronizer implements Runnable{private static final long serialVersionUID = 6138294804551838833L;final Thread thread;Runnable firstTask;volatile long completedTasks;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}//其余代码...
}

Worker类实现了Runnable接口，所以Worker也是一个线程任务。在构造方法中，创建了一个线程，线程的任务就是自己。所以addWorker方法源码下半部分中的第4步t.start，会触发Worker类的run方法中的runWorker被JVM调用。

接着我们再看看runWorker的源码逻辑：

// Worker.runWorker方法源代码
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;// 1.线程启动之后，通过unlock方法释放锁w.unlock(); // allow interruptsboolean completedAbruptly = true;try {// 2.Worker执行firstTask或从workQueue中获取任务，如果getTask方法不返回null,循环不退出while (task != null || (task = getTask()) != null) {// 2.1进行加锁操作，保证thread不被其他线程中断（除非线程池被中断）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// 2.2检查线程池状态，倘若线程池处于中断状态，当前线程将中断。 if ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {// 2.3执行beforeExecute beforeExecute(wt, task);Throwable thrown = null;try {// 2.4执行任务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 {// 2.5执行afterExecute方法 afterExecute(task, thrown);}} finally {task = null;w.completedTasks++;// 2.6解锁操作w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}
}

首先去执行创建这个worker时就有的任务，当执行完这个任务后，worker的生命周期并没有结束，在while循环中，worker会不断地调用getTask方法从阻塞队列中获取任务然后调用task.run()执行任务,从而达到复用线程的目的。只要getTask方法不返回null,此线程就不会退出。

当然，核心线程池中创建的线程想要拿到阻塞队列中的任务，先要判断线程池的状态，如果STOP或者TERMINATED，返回null。

最后看看getTask方法的实现:

// Worker.getTask方法源码
private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();int rs = runStateOf(c);// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?// 1.allowCoreThreadTimeOut变量默认是false,核心线程即使空闲也不会被销毁// 如果为true,核心线程在keepAliveTime内仍空闲则会被销毁。 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;// 2.如果运行线程数超过了最大线程数，但是缓存队列已经空了，这时递减worker数量。 // 如果有设置允许线程超时或者线程数量超过了核心线程数量，// 并且线程在规定时间内均未poll到任务且队列为空则递减worker数量if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {// 3.如果timed为true(想想哪些情况下timed为true),则会调用workQueue的poll方法获取任务.// 超时时间是keepAliveTime。如果超过keepAliveTime时长，// poll返回了null，上边提到的while循序就会退出，线程也就执行完了。// 如果timed为false（allowCoreThreadTimeOut为falsefalse// 且wc > corePoolSize为false），则会调用workQueue的take方法阻塞在当前。// 队列中有任务加入时，线程被唤醒，take方法返回任务，并执行。Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}

核心线程的会一直卡在workQueue.take方法，被阻塞并挂起，不会占用CPU资源，直到拿到Runnable 然后返回（当然如果allowCoreThreadTimeOut设置为true,那么核心线程就会去调用poll方法，因为poll可能会返回null,所以这时候核心线程满足超时条件也会被销毁）。

非核心线程会workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ，如果超时还没有拿到，下一次循环判断compareAndDecrementWorkerCount就会返回null,Worker对象的run()方法循环体的判断为null,任务结束，然后线程被系统回收。

三、四种常见的线程池

Executors类中提供的几个静态方法来创建线程池。大家到了这一步，如果看懂了前面讲的ThreadPoolExecutor构造方法中各种参数的意义，那么一看到Executors类中提供的线程池的源码就应该知道这个线程池是干嘛的。

1. newCachedThreadPool

public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
}

CacheThreadPool的运行流程如下：

提交任务进线程池。
因为corePoolSize为0的关系，不创建核心线程，线程池最大为Integer.MAX_VALUE。
尝试将任务添加到SynchronousQueue队列。
如果SynchronousQueue入列成功，等待被当前运行的线程空闲后拉取执行。如果当前没有空闲线程，那么就创建一个非核心线程，然后从SynchronousQueue拉取任务并在当前线程执行。
如果SynchronousQueue已有任务在等待，入列操作将会阻塞。

当需要执行很多短时间的任务时，CacheThreadPool的线程复用率比较高，会显著的提高性能。而且线程60s后会回收，意味着即使没有任务进来，CacheThreadPool并不会占用很多资源。

2. newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}

核心线程数量和总线程数量相等，都是传入的参数nThreads，所以只能创建核心线程，不能创建非核心线程。因为LinkedBlockingQueue的默认大小是Integer.MAX_VALUE，故如果核心线程空闲，则交给核心线程处理；如果核心线程不空闲，则入列等待，直到核心线程空闲。

与CachedThreadPool的区别：

因为 corePoolSize == maximumPoolSize ，所以FixedThreadPool只会创建核心线程。而CachedThreadPool因为corePoolSize=0，所以只会创建非核心线程。
在 getTask() 方法，如果队列里没有任务可取，线程会一直阻塞在 LinkedBlockingQueue.take() ，线程不会被回收。 CachedThreadPool会在60s后收回。
由于线程不会被回收，会一直卡在阻塞，所以没有任务的情况下， FixedThreadPool占用资源更多。
都几乎不会触发拒绝策略，但是原理不同。FixedThreadPool是因为阻塞队列可以很大（最大为Integer最大值），故几乎不会触发拒绝策略；CachedThreadPool是因为线程池很大（最大为Integer最大值），几乎不会导致线程数量大于最大线程数，故几乎不会触发拒绝策略。

3. newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));
}

有且仅有一个核心线程（ corePoolSize == maximumPoolSize=1），使用了LinkedBlockingQueue（容量很大），所以，不会创建非核心线程。所有任务按照先来先执行的顺序执行。如果这个唯一的线程不空闲，那么新来的任务会存储在任务队列里等待执行。

4. newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {return new ScheduledThreadPoolExecutor(corePoolSize);
}//ScheduledThreadPoolExecutor():
public ScheduledThreadPoolExecutor(int corePoolSize) {super(corePoolSize, Integer.MAX_VALUE,DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,new DelayedWorkQueue());
}

四种常见的线程池基本够我们使用了，但是在《阿里把把开发手册》不建议我们直接使用Executors类中的线程池，而是通过ThreadPoolExecutor的方式，这样的处理方式让写的同学需要更加明确线程池的运行规则，规避资源耗尽的风险。

但如果你及团队本身对线程池非常熟悉，又确定业务规模不会大到资源耗尽的程度（比如线程数量或任务队列长度可能达到Integer.MAX_VALUE）时，其实是可以使用JDK提供的这几个接口的，它能让我们的代码具有更强的可读性。

总结

讲到这里本章对多线程系列线程池知识讲解也就结束了，如果想了解更多知识可以在对应的专栏中看系列文章，谢谢大家的观看，希望能给各位同学带来帮助。如果觉得博主写的还可以的，可以点赞收藏。

侠客岛--多线程系列之线程池(十二)相关推荐

Java多线程系列--“JUC线程池”06之 Callable和Future
转载自 Java多线程系列--"JUC线程池"06之 Callable和Future Callable 和 Future 简介 Callable 和 Future 是比较有趣的一 ...
多线程之 ForkJoinool线程池（二十四）
"分而治之"的一个有效的处理大数据的方法,著名的MapReduce就是采用这种分而治之的思路,简单点生活如果要处理的1000个数据,但是我们不具备处理1000个数据的能量,可以处理 ...
Java多线程系列 JUC线程池05 线程池原理解析(四)
转载 http://www.cnblogs.com/skywang12345/p/3544116.html https://blog.csdn.net/programmer_at/article/d ...
Java多线程系列 JUC线程池01 线程池框架
转载 http://www.cnblogs.com/skywang12345/p/3509903.html 为什么引入Executor线程池框架 new Thread()的缺点 1. 每次new T ...
linux下c语言线程传参数,【linux】C语言多线程中运行线程池,在线程池中运行线程池,,传递的结构体参数值为空/NULL/0...
C语言多线程中运行线程池,在线程池中运行线程池,,传递的结构体参数值为空/NULL/0 本贴问题,之前已经提问过一次,当时已经解决了,原贴在这里https://segmentfault.com/q/1 ...
小豹子带你看源码：Java 线程池（二）实例化
承上启下:上一篇文章小豹子讲了我为什么想要研究线程池的代码,以及我计划要怎样阅读代码.这篇文章我主要阅读了线程池实例化相关的代码,并提出了自己的疑问. 3 千里之行,始于实例化 3.1 先创建一个线程 ...
Android线程和线程池（二）HandlerThread学习记录使用+源码
HandlerThreadAndroid线程和线程池(二)HandlerThread学习记录使用+源码一.作用二.工作原理三.HandlerThread的特点优势: 劣势: 四.使用五.源 ...
Python 线程池 ThreadPoolExecutor(二) - Python零基础入门教程
目录一.Python 线程池前言二.Python 线程池 ThreadPoolExecutor 常用函数 1.线程池 as_completed 函数使用 2.线程池 map 函数使用 3.线程池 ...
多线程编程学习笔记——线程池（二）
接上文多线程编程学习笔记--线程池(一) 三.线程池与并行度此示例是学习如何应用线程池实现大量的操作,及与创建大量线程进行工作的区别. 1. 代码如下 using System; using Sy ...

侠客岛--多线程系列之线程池(十二)