【并发编程】线程池及Executor框架

文章目录

1.为什么要使用线程池
2.线程池创建线程
3.ThreadPoolExecutor类
4.深入剖析线程池实现原理
5.线程池使用示例

1.为什么要使用线程池

   诸如 Web 服务器、数据库服务器、文件服务器或邮件服务器之类的许多服务器应用程序都面向处理来自某些远程来源的大量短小的任务。请求以某种方式到达服务器，这种方式可能是通过网络协议（例如 HTTP、FTP ）、通过 JMS队列或者可能通过轮询数据库。不管请求如何到达，服务器应用程序中经常出现的情况是：单个任务处理的时间很短而请求的数目却是巨大的。每当一个请求到达就创建一个新线程，然后在新线程中为请求服务，但是频繁的创建线程，销毁线程所带来的系统开销其实是非常大的。

    线程池为线程生命周期开销问题和资源不足问题提供了解决方案。通过对多个任务重用线程，线程创建的开销被分摊到了多个任务上。其好处是，因为在请求到达时线程已经存在，所以无意中也消除了线程创建所带来的延迟。这样，就可以立即为请求服务，使应用程序响应更快。而且，通过适当地调整线程池中的线程数目，也就是当请求的数目超过某个阈值时，就强制其它任何新到的请求一直等待，直到获得一个线程来处理为止，从而可以防止资源不足。

风险与机遇:
用线程池构建的应用程序容易遭受任何其它多线程应用程序容易遭受的所有并发风险，
诸如同步错误和死锁，它还容易遭受特定于线程池的少数其它风险，诸如与池有关的死锁、资源不足和线程泄漏。

2.线程池创建线程

Java通过Executors提供四种线程池
- newCachedThreadPool创建一个可缓存的线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
- newFixedThreadPool创建一个定长的线程池，可以控制线程的最大并发数，超出的线程会在队列中等待。
- newScheduledThreadPool创建一个定长的线程池，支持定时及周期性任务执行。
- newSingleThreadExecutor创建一个单线程的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
线程代码

public class ThreadForPools implements Runnable {private Integer index;public ThreadForPools(Integer index) {this.index = index;}@Overridepublic void run() {try {System.out.println("开始处理线程");Thread.sleep(index*100);System.out.println("我的线程标识是"+this.toString());} catch (InterruptedException e) {e.printStackTrace();}}
}

newCachedThreadPool
可以有无限多的线程进来（线程地址不一样），但是需要注意机器的性能。

/*** 创建可缓存的线程池*/
public class MyCachedThreadPool {public static void main(String[] args) {ExecutorService cachedThreadPool = Executors.newCachedThreadPool();for (int i = 0; i < 5; i++) {cachedThreadPool.execute(new ThreadForPools(i));}}
}

newFixedThreadPool
每次最多只有指定个线程在处理，当第一批线程执行完毕后，新的线程进来进行处理（线程地址不一样）。

public class MyFixedThreadPool {public static void main(String[] args) {//线程池允许同时存在两个线程ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);for (int i = 0; i < 5; i++) {fixedThreadPool.execute(new ThreadForPools(i));}}
}

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

//schedule(commod,delay,unit)这个方法是说明系统启动后，需要等待多久时间执行，delay是等待的时间，只执行一次没有周期性。public class MyScheduledThreadPool {public static void main(String[] args) {ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);for (int i = 0; i < 5; i++) {scheduledThreadPool.schedule(new ThreadForPools(i),5,TimeUnit.SECONDS);}}
}

//scheduleAtFixedRate(commod,initialDelay,period,unit),这个是以period周期性执行任务，initialDelay是系统启动等待时间。public class MyScheduledThreadPool {public static void main(String[] args) {ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);for (int i = 0; i < 5; i++) {//线程启动等待2秒执行，之后每3秒执行一个周期scheduledThreadPool.scheduleAtFixedRate(new ThreadForPools(i),2,3,TimeUnit.SECONDS);}}
}

//scheduleWithFixedDelay(commod,initialDelay,period,unit),这个是以period周期性执行任务，initialDelay是系统启动等待时间,和scheduleAtFixedRate的区别在于系统等待的时间不记在周期性执行的时间内。public class MyScheduledThreadPool {public static void main(String[] args) {ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);for (int i = 0; i < 5; i++) {//线程启动等待2秒执行，之后每3秒执行一个周期scheduledThreadPool.scheduleWithFixedDelay(new ThreadForPools(i),5,3,TimeUnit.SECONDS);}}
}

newSingleThreadExecutor
创建一个单线程化的线程池，他只会用唯一的工作线程来执行任务，保证所有任务按照顺序（FIFO、LIFO、优先级）执行。

public class MySingleThreadExecutor {public static void main(String[] args) {ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();for (int i = 0; i < 5; i++) {singleThreadExecutor.execute(new ThreadForPools(i));}}
}

3.ThreadPoolExecutor类

java.util.concurrent.ThreadPoolExecutor类时线程池中最核心的一个类，因此如果要彻底了解java中的线程池，必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。

public class ThreadPoolExecutor extends AbstractExecutorService {.....public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);...
}

从上面的代码可以得知，ThreadPoolExecutor继承了AbstractExecutorService类，并提供了四个构造器，事实上，通过观察每个构造器的源码具体实现，发现前面三个构造器都是调用的第四个构造器进行的初始化工作。
下面解释下各个参数的含义：
corePoolSize：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这两个方法的名字就可以看出，是预创建线程的意思，记载没有任务到来之前就创建corePoolSize个线程或一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把达到的任务方到缓存队列中。
maximumPoolSize：线程池中最大的线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多个个线程。
keepAliveTime：表示线程没有任务执行时最多能保持多久时间会终止。默认情况下当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，keepAliveTime就不会起作用，即不会对初始化预创建的线程起作用。但是如果调用了allowCoreThreadTimeOut(boolean)方法，keepAliveTime也会对预创建的线程起作用，直至线程池中线程为0。
unit：参数keepAliveTime的时间单位，有7种取值
- TimeUnit.DAYS; //天
- TimeUnit.HOURS; //小时
- TimeUnit.MINUTES; //分钟
- TimeUnit.SECONDS; //秒
- TimeUnit.MILLISECONDS; //毫秒
- TimeUnit.MICROSECONDS; //微妙
- TimeUnit.NANOSECONDS; //纳秒
workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：
- ArrayBlockingQueue
- LinkedBlockingQueue
- SynchronousQueue
threadFactory：线程工厂，主要用来创建线程。
handler：表示当拒绝处理任务时的策略，有以下四种取值：
- ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
- ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
- ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
- ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
类关系图

Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；然后ThreadPoolExecutor继承了类AbstractExecutorService。

在ThreadPoolExecutor类中有几个非常重要的方法：
- execute()
- submit()
- shutdown()
- shutdownNow()
execute()：实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。
submit()：是ExecutorService中声明的方法，在AbstractExecutorService就已经由了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法是用来向线程池中提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用execute()方法，只不过他利用了Future来获取任务的执行结果。
shutdown()和shutdownNow()是用来关闭线程池的。

4.深入剖析线程池实现原理

（1）线程池的状态

volatile int runState;static final int RUNNING = 0;static final int SHUTDOWN = 1;static final int stop = 2;static final int TERMINATED = 3;

runState表示当前线程池的状态，用volatile变量用来保证线程之间的可见性。
当创建线程池后，初始化时为RUNNING状态。
如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时它不能够接受新的任务，它会等所有的任务执行完毕。
如果调用了shutdownNow()方法，则线程池处于SHOP状态，此时线程池不能接受新的任务，并且去尝试终止正在运行的任务。
当线程池处于SHUTDOWN或者STOP状态，并且所有工作线程已经很销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATEDzhua状态

（2）任务的执行

ThreadPoolExecutor类的核心成员：

private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务private final ReetrantLock mainLock  = new ReetrantLock(); //线程池的主要锁状态，对线程池的状态改变的核心锁private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集private volatile long keepAliveTime; //线程存活时间private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间private volatile int corePoolSize; //核心池的大小（即线程池中线程数目大于这个参数时，提交的任务会放在任务缓存队列）private volatile int maximumPoolSize; //线程池中最大能容忍的线程数private volatile int poolSize; //线程池中当前线程数private volatile RejectedExecutionHandler handler; //任务拒绝策略private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程private int largestPoolSize; //用来记录线程池中曾经出现的最大线程数private long completedTaskCount; //用来记录已经执行完毕的任务个数

在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法，所以我们只需要研究execute()方法的实现原理即可：

public void execute(Runnable command) {//首先先判断传入的任务是否为空，若是null，则抛出空指针异常；if (command == null)throw new NullPointerException();//如果当前线程数量不小于核心池的数量或者执行addIfUnderCorePoolSize()方法，返回false主席那个下面代码块if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {//判断线程池状态是否为RUNNING，并且放入缓存队列if (runState == RUNNING && workQueue.offer(command)) {//这句话是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用了shutdown或者shutdownNow方法时，那就调用//ensureQueuedTaskHandled方法if (runState != RUNNING || poolSize == 0)ensureQueuedTaskHandled(command);}//如果不是RUNNING状态，并且调用addIfUnderMaximumPoolSize方法失败，则执行拒绝处理。else if (!addIfUnderMaximumPoolSize(command))//拒绝处理reject(command); // is shutdown or saturated}
}

（3）线程池中的线程初始化

默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务后才会创建线程。
在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个办法办到：
- prestartCoreThread()：初始化一个核心线程
- prestartAllCoreThreads()：初始化所有核心线程

public boolean prestartCoreThread(){return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}public int prestartAllCoreThreads(){int n = 0;while(addIfUnderCorePoolSize(null)) //注意传进去的参数是null++n;return n;
}

注意上面传进去的参数null，r = workQueue.take()，即等待任务队列中有任务。

（4）任务缓存队列以及排队策略

workQueue的类型为BlockingQueue<Runnable> ,通常可以取下面三种类型：
- ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小。
- LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE。
- synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。

（5）任务拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常以下四种策略：

ThreadPoolExecutor.AbortPolicy：丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

（6）线程池的关闭

ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()，其中：

shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

（7）线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()

setCorePoolSize：设置核心池大小
setMaximumPoolSize：设置线程池最大能创建的线程数目大小
当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务

5.线程池使用示例

public class ThreadPoolExecutorDemo {public static void main(String[] args) {//创建线程池，核心池5个，最大的线程池数量10个，多余线程空闲存活时间，任务缓存对列及排队策略ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(5));for (int i = 0; i < 15; i++) {//创建15个任务MyTask myTask = new MyTask(i);//每创建一个放在线程池中executor.execute(myTask);/*** getPoolSize():获取线程池中线程数目* getQueue().size():获取队列中等待的任务数目* getCompletedTaskCount():获取已经执行完成的任务数目*/System.out.println("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" +executor.getQueue().size() + "，已执行完的任务数目：" + executor.getCompletedTaskCount());}//结束线程池生命周期executor.shutdown();}
}class MyTask implements Runnable {private int taskNum;public MyTask(int taskNum) {this.taskNum = taskNum;}@Overridepublic void run() {System.out.println("正在执行task：" + taskNum);try {Thread.currentThread().sleep(4000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("task：" + taskNum + "执行完毕");}
}

【并发编程】线程池及Executor框架相关推荐

Java 并发编程 -- 线程池源码实战
一.概述小编在网上看了好多的关于线程池原理.源码分析相关的文章,但是说实话,没有一篇让我觉得读完之后豁然开朗,完完全全的明白线程池,要么写的太简单,只写了一点皮毛,要么就是是晦涩难懂,看完之后几乎都 ...
java线程池_Java 并发编程线程池源码实战
作者 | 马启航杏仁后端工程师.「我头发还多,你们呢?」一.概述笔者在网上看了好多的关于线程池原理.源码分析相关的文章,但是说实话,没有一篇让我觉得读完之后豁然开朗,完完全全的明白线程池,要么写 ...
Java多线程学习（八）线程池与Executor 框架
历史优质文章推荐: Java并发编程指南专栏分布式系统的经典基础理论可能是最漂亮的Spring事务管理详解面试中关于Java虚拟机(jvm)的问题看这篇就够了目录: [TOC] 本节思维导图: ...
一文搞懂线程池原理——Executor框架详解
文章目录 1 使用线程池的好处 2 Executor 框架 2.1 Executor 框架结构 2.2 Executor 框架使用示意图 2.3 Executor 框架成员 2.3.1 Executo ...
java workerdone_【架构】Java并发编程——线程池的使用
前言如果我们要使用线程的时候就去创建一个,这样虽然非常简便,但是就会有一个问题: 如果并发的线程数量很多,并且每个线程都是执行一个时间很短的任务就结束了,这样频繁创建线程就会大大降低系统的效率,因为 ...
Java并发编程——线程池的使用
在前面的文章中,我们使用线程的时候就去创建一个线程,这样实现起来非常简便,但是就会有一个问题: 如果并发的线程数量很多,并且每个线程都是执行一个时间很短的任务就结束了,这样频繁创建线程就会大大降低系统 ...
java并发编程——线程池的工作原理与源码解读
2019独角兽企业重金招聘Python工程师标准>>> 线程池的简单介绍基于多核CPU的发展,使得多线程开发日趋流行.然而线程的创建和销毁,都涉及到系统调用,比较消耗系统资源,所以 ...
java并发测试线程池,Java并发编程——线程池
1.任务与执行策略间的隐性耦合一些任务具有这样的特征:需要或者排斥某种特定的执行策略.对其他任务具有依赖性的任务,就会要求线程池足够大,来保证它锁依赖任务不必排队或者不被拒绝:采用线程限制的任务需要 ...
并发编程--线程池原理
阻塞队列和非阻塞队列 ConcurrentLinkedQueue类适用于高并发场景下的队列,通过无锁的方式,实现了高并发状态下的高性能,通常ConcurrentLinkedQueue性能好于Bloc ...