线程池

作用
用法
- 建议设定大小
- 快捷构造线程池
- submit与execute
- shutdown与shutdownNow
- Future与FutureTast
- 代码
状态
底层原理
- 继承关系
- 主要参数
- 工作原理
- 饱和策略
- 连接复用

作用

1，降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗
2，提高响应速度。当任务到达时，任务可以不需要的等到线程创建就能立即执行
3，提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控

尽量通过Executor来启动线程，这种方法比使用Thread的start方法更好，除了更易管理，效率更好（用线程池实现，节约开销）外，还有关键的一点：有助于避免 this 逃逸问题。

用法

使用ThreadPoolExecutor构造线程池，将线程（任务）传入

创建线程池时需要设定特殊参数，如核心线程池大小、最大线程池大小、缓冲队列大小等

获取结果也是线程池使用的一大难点，普通的Future接口以及FutureTask、ListenableFuture等都可以实现接受结果

建议设定大小

1、CPU计算较多的时候，被叫做CPU密集型应用，核心线程数设置为N+1，N为CPU个数
2、IO操作较多时，被叫做IO密集型应用，设置为2*N

那么这个大小到底是如何确定下来的呢？

通过Little’s Law（利特尔法则）确定的，这个法则说一个系统请求数等于请求的到达率与平均每个单独请求花费的时间之乘积，也就是说知道三点即可确定线程数：请求CPU处理速度、一个请求所消耗的时间、CPU数目

以上过程比较理论化，在实战中会有很多特殊情况发生，比如下午3，4点的流量，和 12 点左右午饭时的流量不一样，而美团给出了动态化配置的解决方案

在源码中有一些诡异的方法，我们一般不会注意到，比如：

    public void setCorePoolSize(int corePoolSize) {if (corePoolSize < 0)throw new IllegalArgumentException();int delta = corePoolSize - this.corePoolSize;this.corePoolSize = corePoolSize;if (workerCountOf(ctl.get()) > corePoolSize)interruptIdleWorkers();else if (delta > 0) {// We don't really know how many new threads are "needed".// As a heuristic, prestart enough new workers (up to new// core size) to handle the current number of tasks in// queue, but stop if queue becomes empty while doing so.int k = Math.min(delta, workQueue.size());while (k-- > 0 && addWorker(null, true)) {if (workQueue.isEmpty())break;}}}

该方法让我们可以在线程池运行时修改核心线程数，JDK不止提供了这个方法，最大线程数等都可以修改。这就叫动态配置

快捷构造线程池

只建议用ThreadPoolExecutor来创建线程池
不建议使用Executors中的以下四种方法创建
前两种队列大小可达INTEGER_VALUE，后两种线程多少可达INTEGER_VALUE，而这两种都会消耗系统资源
在源码中以下方法也只是返回参数固定的ThreadPoolExecutor对象

1，FixedThreadPool：固定线程数的线程池
2，SingleThreadExecutor：只有一个线程的线程池
3，CachedThreadPool：主线程提交任务的速度高于线程处理任务的速度时，会不断创建新的线程
4，ScheduledThreadPoolExecutor：定时执行任务

submit与execute

submit可以提交Callable子类对象并获得一个Future类型的对象，比如FutrueTask作为返回值。可以通过Future的get方法来获取返回值，不过get会阻塞调用该方法的线程，因此是同步的，get方法可以带时间，时间一过就会抛出异常

execute方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否，同时由于Runnable的run方法没有抛出异常，因此Runnable实现的线程也不会抛出异常

shutdown与shutdownNow

shutdown：关闭线程池，线程池的状态变为SHUTDOWN，不再接受新任务，并执行所有在队列中的任务

shutdownNow：立即关闭线程池，线程池的状态变为STOP，停止执行所有任务，返回在队列中的任务链表

同时，还有isShutdown()方法与isTerminated()方法来判断它是否执行shutdown方法以及是否抛出了所有队列进入了terminated状态

Future与FutureTast

Future是一个接口，里面定义了一些方法：

boolean cancel(boolean mayInterruptIfRunning);如果任务已经启动，执行cancel(true)方法将以中断执行此任务线程的方式来试图停止任务
boolean isCancelled();
boolean isDone();
V get();
V get(long timeout, TimeUnit unit);

FutureTask除了实现了Future接口外还实现了Runnable接口（即可以通过Runnable接口实现线程，也可以通过Future取得线程执行完后的结果），因此FutureTask也可以直接提交给Executor执行

代码

public class test4 {private static final int CORE_POOL_SIZE = 6;private static final int MAX_POOL_SIZE = 10;private static final int QUEUE_CAPACITY = 100;private static final Long KEEP_ALIVE_TIME = 1L;public static void main(String[] args) throws InterruptedException {TreadTest t = new TreadTest();ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE,MAX_POOL_SIZE,KEEP_ALIVE_TIME,TimeUnit.SECONDS,new ArrayBlockingQueue<>(QUEUE_CAPACITY),new ThreadPoolExecutor.CallerRunsPolicy());for(int i = 0; i < 30; i++) {//          TreadTest t = new TreadTest();executor.submit(t);}executor.shutdown();while (!executor.isTerminated()) {}System.out.println("Finished all threads");}
}class TreadTest implements Callable<Integer> {volatile int j = 0;@Overridepublic Integer call() {for(int i = 0; i < 10; i++) {System.out.println("Thread Name= "+Thread.currentThread().getName() + "number = "+ j++);try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}//formatter pattern is changed here by thread, but it won't reflect to other threadsreturn j;}
}

状态

线程池有5种状态

1，RUNNING：线程池一旦被创建，就处于 RUNNING 状态，任务数为0，能够接收新任务，对已排队的任务进行处理。

2，SHUTDOWN：不接收新任务，但能处理已排队的任务。调用线程池的shutdown()方法，线程池由RUNNING转变为SHUTDOWN状态。

3，STOP：不接收新任务，不处理已排队的任务，并且会中断正在处理的任务。调用线程池的shutdownNow()方法，线程池由RUNNING或SHUTDOWN转变为STOP状态。

4，TIDYING（整理）：

SHUTDOWN状态下，任务数为 0，其他所有任务已终止，线程池会变为TIDYING状态，会执行terminated方法。线程池中的terminated()方法是空实现，可以重写该方法进行相应的处理
线程池在SHUTDOWN状态，任务队列为空且执行中任务为空，线程池就会由SHUTDOWN转变为TIDYING状态
线程池在STOP状态，线程池中执行中任务为空时，就会由STOP转变为TIDYING状态

5，TERMINATED（结束）：线程池彻底终止。线程池在TIDYING状态执行完terminated()方法就会由TIDYING转变为TERMINATED状态

底层原理

继承关系

Executor接口 ->
ExecutorServer接口 ->
AbstractExecutorServer接口 ->
ThreadPoolExecutor类
如果加入已计划的线程池，就成了如下结果：

ScheduledExecutorService主要是用来做定时任务的

以下是ExecutorService接口的所有方法（Execute接口只有一个execute方法）

可以看到该接口就是表示了线程池的生命周期，这个线程池服务的invokeAll方法是将集合中所有的Callable都执行，invokeAny提交所有但是只返回一个最先完成的结果，其他的主要方法下面会说

以下源码主要来自ThreadPoolExecutor类

主要参数

corePoolSize：核心线程数，线程池有多少线程同时运行

maximumPoolSize：最大线程数，当提交的任务超过队列大小时，当前可以同时运行的线程数量变为最大线程数

workQueue：缓冲队列，指最大可以存放的任务数，注意，线程执行任务时任务还在队列中，可以把缓冲队列设置为0看看效果

handler：饱和策略，如果缓冲队列慢了会怎么处理提交的任务

工作原理

1，核心线程数未满时，提交任务，无可用线程时线程数加一
2，核心线程数已满时，提交任务，到达缓冲队列
3，缓冲队列已满时，提交任务，创建新线程直到到达最大线程数
4，继续提交任务，线程池使用饱和策略，饱和策略可以在构造线程池时设定

以上四步也是ThreadPoolExecutor的execute方法的过程

但是在开始之前需要来一点预备知识

 // ctl 表示了线程的状态以及当前激活的线程数，用一个值表示了两种东西，很离谱对不对private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));// Integer.SIZE 就是32，COUNT_BITS 就是29private static final int COUNT_BITS = Integer.SIZE - 3;private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 这里就是线程池运行时状态private static final int RUNNING    = -1 << COUNT_BITS;private static final int SHUTDOWN   =  0 << COUNT_BITS;private static final int STOP       =  1 << COUNT_BITS;private static final int TIDYING    =  2 << COUNT_BITS;private static final int TERMINATED =  3 << COUNT_BITS;// 提供读取当前线程数、当前运行状态的方法private static int runStateOf(int c)     { return c & ~CAPACITY; }private static int workerCountOf(int c)  { return c & CAPACITY; }private static int ctlOf(int rs, int wc) { return rs | wc; }

来解释一下，ctl 的低29位用于存放当前的线程数，因此一个线程池在理论上最大的线程数是(2^29)-1；高3位是用于表示当前线程池的状态，其中高三位的值和状态对于如下：

111: RUNNING
000: SHUTDOWN
001: STOP
010: TIDYING
011: TERMINATED

在以后的处理需要多个数据的问题的时候，也可以模仿这种优雅的写法。接下来来看看execute方法

    public void execute(Runnable command) {if (command == null)throw new NullPointerException();// ctl 是个原子类，拿到ctl的值int c = ctl.get();// 如果核心线程数大于现在正在执行的线程数，workerCountOf方法用于获取当前正在执行的线程数if (workerCountOf(c) < corePoolSize) {// 调用addWorker创建一个线程并返回，如果创建失败再获取一个ctlif (addWorker(command, true))return;c = ctl.get();}// 如果当前线程池在跑并且将command成功加入了队列if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();// 如果当前线程池没在跑并且将command删除成功了，则执行拒绝策略if (! isRunning(recheck) && remove(command))reject(command);// 如果发生什么事情删除失败了或者当前线程池在运行中，则会判断工作线程是否为0 ，如果过为0 就创建一个非核心线程else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 尝试创建一个工作线程，如果线程池挂了或者大于最大线程数，执行拒绝策略else if (!addWorker(command, false))reject(command);// 其他情况不做处理}

其中addWorker方法用于创建新线程，它使用HashSet来存放线程，里面放的是Worker，只有持有全局锁mainLock的前提下才能访问此集合。同时这个方法返回true则表示线程创建成功，false表示失败

    private final ReentrantLock mainLock = new ReentrantLock();// 线程池的最大大小private int largestPoolSize;// 工作线程放在这里private final HashSet<Worker> workers = new HashSet<>();private boolean addWorker(Runnable firstTask, boolean core) {//这个retry用于跳出两层循环，两次循环创建失败后再次进行资格判断retry:for (;;) {//条件判断int c = ctl.get();int rs = runStateOf(c);if (rs >= SHUTDOWN &&! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))return false;//以下for循环主要为了将workcount的数量加1for (;;) {//获取线程池中工作的线程的数量int wc = workerCountOf(c);// core参数为true的话表明队列也满了，线程池大小变为 maximumPoolSizeif (wc >= CAPACITY ||wc >= (core ? corePoolSize : maximumPoolSize))return false;//原子操作将workcount的数量加1if (compareAndIncrementWorkerCount(c))break retry;// 如果线程的状态改变了就再次执行上述操作c = ctl.get();if (runStateOf(c) != rs)continue retry;}}// 标记工作线程是否启动成功boolean workerStarted = false;// 标记工作线程是否创建成功boolean workerAdded = false;Worker w = null;//以下过程尝试去创建工作线程try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {// 加锁final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {//获取线程池状态int rs = runStateOf(ctl.get());//rs < SHUTDOWN 如果线程池状态依然为RUNNING,并且线程的状态是存活的话，就会将工作线程添加到工作线程集合中//(rs=SHUTDOWN && firstTask == null)如果线程池状态小于STOP，也就是RUNNING或者SHUTDOWN状态下，同时传入的任务实例firstTask为null，则需要添加到工作线程集合和启动新的Worker// firstTask == null证明只新建线程而不执行任务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();}// 如果成功添加工作线程，则调用Worker内部的线程实例t的Thread#start()方法启动真实的线程实例if (workerAdded) {t.start();// 标记线程启动成功workerStarted = true;}}} finally {// 线程启动失败，需要从工作线程中移除对应的Workerif (! workerStarted)addWorkerFailed(w);}return workerStarted;}

饱和策略

1，拒绝执行任务并抛出异常
2，使用调用线程池的线程执行任务
3，丢弃此任务
4，丢弃第一个等待的任务
还有更多处理方法，可以自行百度

连接复用

private Runnable getTask() {boolean timedOut = false;for (;;) {int c = ctl.get();int wc = workerCountOf(c);// wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量；// 对于超过核心线程数量的这些线程，需要进行超时控制boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if (timed && timedOut) {// 如果需要进行超时控制，且上次从缓存队列中获取任务时发生了超时，那么尝试将workerCount减1,即当前活动线程数减1，// 如果减1成功，则返回null，这就意味着runWorker()方法中的while循环会被退出，其对应的线程就要销毁了，也就是线程池中少了一个线程了if (compareAndDecrementWorkerCount(c))return null;continue;}try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();// 注意workQueue中的poll()方法与take()方法的区别// poll方式取任务的特点是从缓存队列中取任务,最长等待keepAliveTime的时长，取不到返回null// take方式取任务的特点是从缓存队列中取任务，若队列为空,则进入阻塞状态，直到能取出对象为止if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}
}

上面的代码实现了连接池的线程复用，以及超过核心线程数的线程如何被销毁，核心线程如何被保存，思路就是把Runnable放进BlockingQueue里，用一定方式调用线程去拿任务

如果当前活动线程数大于核心线程数，当去缓存队列中取任务的时候，如果缓存队列中没任务了，则等待keepAliveTime的时长，此时还没任务就返回null，这就意味着runWorker()方法中的while循环会被退出，其对应的线程就要销毁了，也就是线程池中少了一个线程了。因此只要线程池中的线程数大于核心线程数就会这样一个一个地销毁这些多余的线程

如果当前活动线程数小于等于核心线程数，同样也是去缓存队列中取任务，但当缓存队列中没任务了，就会进入阻塞状态，直到能取出任务为止，因此这个线程是处于阻塞状态的，并不会因为缓存队列中没有任务了而被销毁。这样就保证了线程池有N个线程是活的，可以随时处理任务，从而达到重复利用的目的

所以最大线程与核心线程的不同，就是调用获取任务的方法不同，一个take一个是poll，这两个方法都由BlockingQueue友情提供

线程池作用、用法以及原理相关推荐

线程池作用和参数原理
线程池的作用减少资源的开销减少了每次创建线程.销毁线程的开销. 提高响应速度每次请求到来时,由于线程的创建已经完成,故可以直接执行任务,因此提高了响应速度. 提高线程的可管理性线程是一种稀缺资 ...
Java 四种线程池的用法分析
1.new Thread的弊端执行一个异步任务你还只是如下new Thread吗? new Thread(new Runnable() {@Overridepublic void run() {// ...
【Android 异步操作】线程池 ( 线程池作用 | 线程池种类 | 线程池工作机制 | 线程池任务调度源码解析 )
文章目录一.线程池作用二.线程池种类三.线程池工作机制四.线程池任务调度源码解析一.线程池作用线程池作用 : ① 避免创建线程 : 避免每次使用线程时 , 都需要创建线程对象 ; ② 统 ...
Qt工作笔记-线程池作用之一：限制系统中执行线程的数量
一.线程池的作用: 线程池作用就是限制系统中执行线程的数量. 根据系统的环境情况,可以自动或手动设置线程数量,达到运行的最佳效果:少了浪费了系统资源,多了造成系统拥挤效率不高.用线程池控制线程数量,其 ...
线程池作用及C语言实现
一.线程池作用 1.减少线程的创建销毁次数,减少性能开销: 当你需要限制你应用程序中同时运行的线程数时,线程池非常有用.因为启动一个新线程会带来性能开销,每个线程也会为其堆栈分配一些内存等.为了任务的 ...
线程池概念、线程池作用、线程池的四种创建方式
线程池 1.1.什么是线程池? 线程池是指在初始化一个多线程应用程序过程中创建一个线程集合,然后在需要执行新的任务时重用这些线程而不是新建一个线程.线程池中线程的数量通常完全取决于可用内存数量和应用程 ...
深入浅出 Java Concurrency (33): 线程池 part 6 线程池的实现及原理 (1)[转]
线程池数据结构与线程构造方法由于已经看到了ThreadPoolExecutor的源码,因此很容易就看到了ThreadPoolExecutor线程池的数据结构.图1描述了这种数据结构. 图1 Thre ...
线程池shutdown和shutdownNow原理和区别
说明:以ThreadPoolExecutor线程池为例说明整个流程(不同的线程池实现上略有差别). 一.shutdown流程 1.流程简介修改线程池状态为SHUTDOWN 不再接收新提交的任务中断 ...
Java 线程池作用及类型
线程池线程池解决了两个不同的问题: 减少线程创建的开销,能提高执行大量异步任务的效率提供了一种限制和管理资源及线程的方法,并且还维护了一些基本的统计信息(如已完成的任务数) 线程池的使用对new ...

线程池作用、用法以及原理

线程池

作用

用法