一文带你了解Java线程池(Executor)-上
Posted Java指南修炼
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一文带你了解Java线程池(Executor)-上相关的知识，希望对你有一定的参考价值。
分析Java线程池就离不开Executor类，今天就让我们来一起好好看下
除开今天要讲的线程池，我还整理了一些技术资料和面试题集，供大家提升进阶，面试突击，不管你是有跳槽打算还是单纯精进自己，都可以免费领取一份。
从面试简历模板到大厂面经汇总，从大厂内部技术资料到互联网高薪必读书单，以及Java面试核心知识点（283页）和Java面试题合集2022年最新版（485页）等等
领取方式在文末！
Executor框架
为了更好地控制多线程，JDK提供了一套Executor框架，可以有效地进行线程控制，其本质上就是一个线程池。
其中ThreadPoolExecutor表示一个线程池。Executors类则扮演着线程池工厂的角色，通过Executors可以取得一个拥有特定功能的线程池。从上图可知，ThreadPoolExecutor类实现了Executor接口，因此，通过这个接口，任何Runnable对象都可以被ThreadPoolExecutor线程池调度。
Executor框架提供了各种类型的线程池，主要有以下工厂方法：
public static ExecutorService newFixedThreadPool(int nThreads)
public static ExecutorService newSingleThreadExecutor()
public static ExecutorService newCachedThreadPool()
public static ScheduledExecutorService newSingleThreadScheduledExecutor()
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)以上工厂方法分别返回具有不同工作特性的线程池。这些线程池工厂方法的具体说明如下：
newFixedThreadPool: 该方法返回一个固定线程数量的线程池。该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂时存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务newSingleThreadExecutor: 该方法返回一个只有一个线程的线程池。若多余一个任务被提交到该线程池，任务会被保存在一个任务队列中，待线程空闲，按先入先出的顺序执行队列中的任务newCachedThreadPool: 该方法返回一个可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用newSingleThreadScheduledExecutor: 该方法返回一个ScheduledExecutorService对象，线程池大小为1。ScheduledExecutorService接口在ExecutorService接口之上扩展了在给定时间执行某任务的功能，如在某个固定的延时之后执行，或者周期性执行某个任务。newScheduledThreadPool: 该方法也返回一个ScheduledExecutorService对象，但该线程池可以指定线程数量
计划任务
一个值得注意的方法是newScheduledThreadPool()。它返回一个ScheduledExecutorService对象，可以根据时间对线程进行调度。它的一些主要方法如下：
public ScheduledFuture<?> schedule(Runnable command,long delay,TimeUnit unit)
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit)
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit)ScheduledExecutorService起到了计划任务的作用，它会在指定的时间，对任务进行调度。
方法schedule()会在给定时间，对任务进行一次调度。方法scheduleAtFixedRate()和scheduleWithFixedDelay()会对任务进行周期性的调度，但是两者有一点区别: 对于FixedRate方式来说，任务调度的频率是一定的。它是以上一个任务开始执行时间为起点，之后的period时间，调度下一次任务；而FixDelay则是在上一个任务结束后，再经过delay时间进行任务调度。
ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
//如果前面的任务没有完成，则调度也不会启动
ses.scheduleAtFixedRate(new Runnable()
@Override
public void run()
try
TimeUnit.SECONDS.sleep(1);
System.out.println(new Date().toLocaleString());
catch (InterruptedException e)
e.printStackTrace();
,0,2,TimeUnit.SECONDS);output：
2017-8-28 21:46:49
2017-8-28 21:46:51
2017-8-28 21:46:53
2017-8-28 21:46:55
2017-8-28 21:46:57
2017-8-28 21:46:59上述输出的单位是秒，可以看到，时间间隔是2秒。如果任务的执行时间改为8秒，会有怎么样的打印
2017-8-28 21:48:54
2017-8-28 21:49:02
2017-8-28 21:49:10
2017-8-28 21:49:18
2017-8-28 21:49:26
2017-8-28 21:49:34可以发现，周期不再是2秒，而是变成了8秒。可知，如果周期太短，那么任务就会在上一个任务结束后，立即被调用。如果改成scheduleWithFixedDelay，并且周期为2秒，任务耗时8秒，那么任务的时间间隔为10秒。
2017-8-28 21:52:20
2017-8-28 21:52:30
2017-8-28 21:52:40
2017-8-28 21:52:50如果任务本身抛出了异常，那么后续的所有执行都会被中断，因此，做好异常处理就非常重要。
ScheduledFuture的使用
ScheduledFuture很简单，它就是在Future基础上还集成了Comparable和Delayed的接口。它用于表示ScheduledExecutorService中提交了任务的返回结果。我们通过Delayed的接口getDelay()方法知道该任务还有多久才被执行。
ScheduledExecutorService service = Executors.newScheduledThreadPool(10);
ScheduledFuture sf = service.schedule(new Callable()
public Object call() throws Exception
System.out.println("job start");
return "ok";
,5, TimeUnit.SECONDS);
TimeUnit.SECONDS.sleep(2);
System.out.println("delay:"+sf.getDelay(TimeUnit.SECONDS));
if(Math.random()>0.5)
System.out.println("and then cancel the job");
sf.cancel(false);//mayInterruptIfRunning : false
else
System.out.println("do not cancel,wait for result:");
System.out.println(sf.get());
service.shutdown();
可以通过cancel来取消一个任务，或者通过get()方法来返回任务的结果(Callable支持，Runnable返回null)
核心线程池的内部实现
对于上面锁列出的线程池，虽然看起来有着完全不同的功能特点，但其内部实现均使用了ThreadPoolExecutor实现，下面给出了这三个线程池的实现方式：
public static ExecutorService newFixedThreadPool(int nThreads)
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
public static ExecutorService newSingleThreadExecutor()
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
public static ExecutorService newCachedThreadPool()
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
由以上线程池的实现代码可以看到，它们都只是ThreadPoolExecutor类的封装，看一下ThreadPoolExecutor最重要的构造函数：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)参数含义如下：
-
corePoolSize指定了线程池中的最小工作线程数量 -
maximumPoolSize指定了线程池中的最大线程数量 -
keepAliveTime当线程池线程数量超过corePoolSize时，多余的空闲线程的存活时间 -
unitkeepAliveTime的单位 -
workQueue任务队列，被提交但尚未被执行的任务 -
threadFactory线程工厂，用于创建线程 -
handler拒绝策略。当任务太多来不及处理，如何拒绝任务
corePoolSize和maximumPoolSize:
线程创建策略如下，通过下面这个流程图可以很好的理解corePoolSize和maximumPoolSize的关系：
来分析一下这个流程图，当一个任务被提交进来后，首先会比较该线程池运行的线程数量与corePoolSize，如果小于(哪怕池中有空闲线程)则实例化一个新线程(来处理这个任务)；否则尝试入队，若入队失败(offer方法返回false)，说明队满，则判断是否小于maximumPoolSize，若小于则新建临时线程；否则执行拒绝策略。
我们可以通过一个实例来验证下这个过程：
import java.util.concurrent.*;
public class ThreadPoolTest
private static class MyTask implements Runnable
private String name;
public MyTask(String name)
this.name = name;
@Override
public String toString()
return name;
@Override
public void run()
System.out.println(Thread.currentThread().getName() + " start handle " + this);
try
Thread.sleep(10000);
System.out.println(Thread.currentThread().getName() + " finished " + this);
catch (InterruptedException e)
e.printStackTrace();
public static void main(String[] args)
//传入了有限队列，大小为5 默认的拒绝策略为抛弃
ExecutorService pool = new ThreadPoolExecutor(2, 4,
0L, TimeUnit.MILLISECONDS,
new TaskQueue(5));
//10个任务
for (int i = 1; i <= 10; i++)
MyTask task = new MyTask("Task-" + i);
try
pool.execute(task);
catch (RejectedExecutionException e)
System.out.println(task + " was rejected.");
//关闭线程池，它会等待已提交的任务执行完毕
pool.shutdown();
/**
* 继承了LinkedBlockingQueue，增加了打印信息
*/
private static class TaskQueue extends LinkedBlockingQueue<Runnable>
public TaskQueue()
super();
public TaskQueue(int capacity)
super(capacity);
@Override
public boolean offer(Runnable runnable)
boolean result = super.offer(runnable);
System.out.println(runnable + " enqueue " + (result ? " success" : "failed."));
return result;
@Override
public Runnable take() throws InterruptedException
Runnable task = super.take();
System.out.println(task + " was finishd and removed.");
return task;
输出如下：
Task-3 enqueue success
Task-4 enqueue success
Task-5 enqueue success
Task-6 enqueue success
Task-7 enqueue success
Task-8 enqueue failed.
Task-9 enqueue failed.
Task-10 enqueue failed.
Task-10 was rejected. //被拒接
pool-1-thread-1 start handle Task-1
pool-1-thread-2 start handle Task-2
pool-1-thread-3 start handle Task-8
pool-1-thread-4 start handle Task-9
pool-1-thread-1 finished Task-1
pool-1-thread-1 start handle Task-3
pool-1-thread-2 finished Task-2
pool-1-thread-2 start handle Task-4
pool-1-thread-3 finished Task-8
pool-1-thread-3 start handle Task-5
pool-1-thread-4 finished Task-9
pool-1-thread-4 start handle Task-6
pool-1-thread-1 finished Task-3
pool-1-thread-1 start handle Task-7
pool-1-thread-2 finished Task-4
pool-1-thread-3 finished Task-5
pool-1-thread-4 finished Task-6
pool-1-thread-1 finished Task-7我们自己实现了一个有界队列，增加了一些打印信息便于理解。构造了一个核心线程数为2，最大线程数为4的线程池。同时，它的有界队列大小为5。也就是说最多能同时运行4个线程，有5个任务在队列中保存，若此时再有任务进来，转而执行拒绝策略。
从上面的输出可以看出，Task-1、Task-2直接被处理，接着Task-3、Task-4、Task-5、Task-6、Task-7入队，然后Task-8、Task-9入队失败，但是此时运行的线程数为2，小于最大的值4，因此这两个任务被新建的临时线程处理；接着Task-10入队失败，同时运行的线程数达到最大值，执行拒绝策略。
还有workQueue和拒绝策略的相关内容我放到下一篇和大家分享，前面说到了为大家准备一份资料，简单介绍下，包含以下内容：
- Java架构师学习路线图（对标阿里P7级别，更高阶的大佬小弟就不在这献丑了）
- 模块化学习资源（Java并发编程、分布式缓存的原理及应用、ZooKeeper原理及应用、Netty网络编程原理及应用、Kafka原理及应用、常见的23种经典设计模式、Spring原理及应用、数据结构与算法……）
- 2022年大厂面试高频知识点整理
资料持续更新中，目前全部都是免费送给大家，如果有需要，尽管拿走，添加我助手领取，备注“CSDN南哥”
Java多线程之线程池 Executor（上）
new Thread的弊端
- 每次new Thread 新建对象，性能差。
- 线程缺乏统一管理，可能无限制的新建线程，相互竞争，可能占用过多的系统资源导致死机或者OOM（out of memory
内存溢出），这种问题的原因不是因为单纯的new一个Thread，而是可能因为程序的bug或者设计上的缺陷导致不断new
Thread造成的。 - 缺少更多功能，如更多执行、定期执行、线程中断
线程池的好处
- 重用存在的线程，减少对象创建、消亡的开销，性能好。
- 可有效控制最大并发线程数，提高系统资源利用率，同时可以避免过多资源竞争，避免阻塞。
- 提供定时执行、定期执行、单线程、并发数控制等功能。
线程池类图
在线程池的类图中，我们最常使用的是最下边的Executors,用它来创建线程池使用线程。那么在上边的类图中，包含了一个Executor框架，它是一个根据一组执行策略的调用调度执行和控制异步任务的框架，目的是提供一种将任务提交与任务如何运行分离开的机制。它包含了三个executor接口：
- Executor:运行新任务的简单接口。
- ExecutorService：扩展了Executor，添加了用来管理执行器生命周期和任务生命周期的方法。
- ScheduleExcutorService：扩展了ExecutorService，支持Future和定期执行任务。
线程池核心类-ThreadPoolExecutor
参数说明：ThreadPoolExecutor一共有七个参数，这七个参数配合起来，构成了线程池强大的功能。
- corePoolSize：核心线程数量
- maximumPoolSize：线程最大线程数
- workQueue：阻塞队列，存储等待执行的任务，很重要，会对线程池运行过程产生重大影响
当我们提交一个新的任务到线程池，线程池会根据当前池中正在运行的线程数量来决定该任务的处理方式。处理方式有三种：
1、直接切换（SynchronusQueue）
2、无界队列（LinkedBlockingQueue）能够创建的最大线程数为corePoolSize,这时maximumPoolSize就不会起作用了。当线程池中所有的核心线程都是运行状态的时候，新的任务提交就会放入等待队列中。
3、有界队列（ArrayBlockingQueue）最大maximumPoolSize，能够降低资源消耗，但是这种方式使得线程池对线程调度变的更困难。因为线程池与队列容量都是有限的。所以想让线程池的吞吐率和处理任务达到一个合理的范围，又想使我们的线程调度相对简单，并且还尽可能降低资源的消耗，我们就需要合理的限制这两个数量
分配技巧：
[如果想降低资源的消耗包括降低cpu使用率、操作系统资源的消耗、上下文切换的开销等等，可以设置一个较大的队列容量和较小的线程池容量，这样会降低线程池的吞吐量。如果我们提交的任务经常发生阻塞，我们可以调整maximumPoolSize。如果我们的队列容量较小，我们需要把线程池大小设置的大一些，这样cpu的使用率相对来说会高一些。但是如果线程池的容量设置的过大，提高任务的数量过多的时候，并发量会增加，那么线程之间的调度就是一个需要考虑的问题。这样反而可能会降低处理任务的吞吐量。
- keepAliveTime：线程没有任务执行时最多保持多久时间终止（当线程中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交核心线程外的线程不会立即销毁，而是等待，直到超过keepAliveTime）
- unit：keepAliveTime的时间单位
- threadFactory：线程工厂，用来创建线程，有一个默认的工场来创建线程，这样新创建出来的线程有相同的优先级，是非守护线程、设置好了名称）
- rejectHandler：当拒绝处理任务时(阻塞队列满)的策略（AbortPolicy默认策略直接抛出异常、CallerRunsPolicy用调用者所在的线程执行任务、DiscardOldestPolicy丢弃队列中最靠前的任务并执行当前任务、DiscardPolicy直接丢弃当前任务）
corePoolSize、maximumPoolSize、workQueue
三者关系：如果运行的线程数小于corePoolSize的时候，直接创建新线程来处理任务。即使线程池中的其他线程是空闲的。如果运行中的线程数大于corePoolSize且小于maximumPoolSize时，那么只有当workQueue满的时候才创建新的线程去处理任务。如果corePoolSize与maximumPoolSize是相同的，那么创建的线程池大小是固定的。这时有新任务提交，当workQueue未满时，就把请求放入workQueue中。等待空线程从workQueue取出任务。如果workQueue此时也满了，那么就使用另外的拒绝策略参数去执行拒绝策略。
初始化方法：由七个参数组合成四个初始化方法
其他方法：
线程池生命周期：
- running：能接受新提交的任务，也能处理阻塞队列中的任务
- shutdown：不能处理新的任务，但是能继续处理阻塞队列中任务
- stop：不能接收新的任务，也不处理队列中的任务
- tidying：如果所有的任务都已经终止了，这时有效线程数为0
- terminated：最终状态
使用Executor创建线程池
使用Executor可以创建四种线程池：分别对应上边提到的四种线程池初始化方法
1、Executors.newCachedThreadPool
创建一个可缓存的线程池，如果线程池的长度超过了处理的需要，可以灵活回收空闲线程。如果没有可回收的就新建线程。
//源码：
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//使用方法：
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(new Runnable() {
@Override
public void run() {
log.info("task:{}", index);
}
});
}
executorService.shutdown();
}
值得注意的一点是，newCachedThreadPool的返回值是ExecutorService类型，该类型只包含基础的线程池方法，但却不包含线程监控相关方法，因此在使用返回值为ExecutorService的线程池类型创建新线程时要考虑到具体情况。
2、newFixedThreadPool
定长线程池，可以线程现成的最大并发数，超出在队列等待
//源码：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//使用方法：
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(new Runnable() {
@Override
public void run() {
log.info("task:{}", index);
}
});
}
executorService.shutdown();
}
3、newSingleThreadExecutor
单线程化的线程池，用唯一的一个共用线程执行任务，保证所有任务按指定顺序执行（FIFO、优先级…）
//源码
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//使用方法：
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
executorService.execute(new Runnable() {
@Override
public void run() {
log.info("task:{}", index);
}
});
}
executorService.shutdown();
}
4、newScheduledThreadPool
定长线程池，支持定时和周期任务执行
//源码：
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,//此处super指的是ThreadPoolExecutor
new DelayedWorkQueue());
}
//基础使用方法：
public static void main(String[] args) {
ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
executorService.schedule(new Runnable() {
@Override
public void run() {
log.warn("schedule run");
}
}, 3, TimeUnit.SECONDS);//延迟3秒执行
executorService.shutdown();
}
ScheduledExecutorService提供了三种方法可以使用：
scheduleAtFixedRate：以指定的速率执行任务
scheduleWithFixedDelay：以指定的延迟执行任务
举例：
executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
log.warn("schedule run");
}
}, 1, 3, TimeUnit.SECONDS);//延迟一秒后每隔3秒执行
小扩展：延迟执行任务的操作，java中还有Timer类同样可以实现
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
log.warn("timer run");
}
}, new Date(), 5 * 1000);
以上是关于一文带你了解Java线程池(Executor)-上的主要内容，如果未能解决你的问题，请参考以下文章