为什么要有线程池
- 加快速度，提高效率(大量的操作需要多个线程进行处理时)
- 达到可复用的目的，减少内存消耗，避免内存问题(大量请求处理例如Tomcat)
- 统一管理
具体核心参数
- corePoolSize 线程池的常驻核心线程数。如果设置为 0，则表示在没有任何任务时，销毁线程池；
- maximumPoolSize 线程池的最大线程数，当到达核心线程数之后，会慢慢向最大线程数变大
- workQueue 线程池执行的任务队列
* sychronizeQueue 同步交换队列 队列大小为1
* LinkedBlockQueue 数据结构为链表的阻塞队列，最大值为Integer.MAX_VALUE
* ArrayBlockQueue 数据结构为数组的阻塞队列，队列长度为设置的值 - keepAliveTime 线程的存活时间,当线程池空闲时并且超过了此时间，多余的线程就会销毁，直到线程池中的线程数量销毁的等于 corePoolSize 为止，如果 maximumPoolSize 等于 corePoolSize，那么线程池在空闲的时候也不会销毁任何线程。
- threadFactory 线程工厂
- RejectedExecutionHandler 线程的拒绝策略，即当任务列已经超过线程数量的最大值之后执行的操作，如下几种策略
- AbortPolicy 后续任务不执行并抛出异常
- DiscardPolicy 超出任务队列的任务不在执行
- DiscardOldestPolicy 丢弃掉队列中最先加入队列的任务，
- CallerRunsPolicy 在线程池的调用线程中执行无法加入队列的任务
- unit 线程存活时间的单位
线程工具类Executors可以创建的几种具体类型得线程池
首先要说明的是几个类之间的关系 Executor是 线程池ThreadExecutorPool执行的最终接口，具体来说是ThreadExecutorPool集成抽象类AbstractExecutorService，抽象类AbstractExecutorService实现 ExecutorService接口， ExecutorService继承Executor接口。然后Executors是创建几种具体线程池的工具类
newFixedThreadPool
使用该静态方法，创建的是固定线程大小的线程池。ExecutorService executorService = Executors.newFixedThreadPool(10);
源码如下
//Executors.java
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
//ThreadPoolExecutor.java
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
线程池的核心线程数和最大线程数都是初始化赋值时的线程数。keepAliveTime为0 表示非核心线程数执行完后立即销毁。但是在newFixedThreadPool 中核心线程数和最大线程数是相同的，即没有非核心线程。所以此处的keepAliveTime为多少都是没有作用的。
另外 此处使用的workQueue为LinkedBlockingQueue，LinkedBlockingQueue的队列最大数值为Integer.MAX_VALUE。因此随着任务队列的数据不断增多，很有可能会导致oom异常
使用如下测试代码进行测试
public class ThreadPool {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i=0;i<Integer.MAX_VALUE;i++){
executorService.execute(new Task());
}
}
}
class Task extends Thread{
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName());
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
vm options设置为-Xmx8m -Xms8m ;可以看到运行不就就会看到OOM异常
file
newCachedThreadPool
newCachedThreadPool从字面意思理解即为缓存的线程池，即是可以利用缓存的线程来执行后续的任务
//Executors.java
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//
可以看到 线程池的几个参数中，核心线程数为0，但是最大线程数为Integer.MAX_VALUE，且任务队列为同步队列，只能容纳一个任务，这样就会导致当任务量太大时创建线程过多会直接OOM异常。但是当任务量小，任务执行不存在阻塞的情况时就可以复用之前缓存的线程.另外使用该线程执行完任务之后，由于核心线程数为0，但是keepAliveTime设置为60s.所以在无任务执行60s之后线程池将会退出。
复用情况
测试代码如下
public class CachedThreadPool {
public static void main(String[] args) {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i=0;i<1000;i++){
cachedThreadPool.execute(new Task());
}
}
}
class Task extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
执行结果
file
可以看到在执行过程中线程存在复用的情况
OOM异常情况
将vm options 设置为-Xmx8m -Xms8m
public class CachedThreadPool {
public static void main(String[] args) {
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i=0;i<Integer.MAX_VALUE;i++){
cachedThreadPool.execute(new Task());
}
}
}
class Task extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
try {
Thread.sleep(5000000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
结果如下图
file
在运行一段时间后出现OOM异常
newSingleThreadExecutor
newSingleThreadExecutor即为线程池中只创建了一个线程的线程池，创建源码如下
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
可以看到核心线程数和最大线程数均设置为1。任务队列使用的是无界的阻塞队列。这也就意味着当大量任务堆积在队列中时可能造成OOM异常
模拟OOM异常情况
public class SingleThreadPool {
public static void main(String[] args) {
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i=0;i<Integer.MAX_VALUE;i++){
singleThreadExecutor.execute(new Task());
}
}
}
class Task extends Thread{
@Override
public void run() {
System.out.println(Thread.currentThread().getName());
try {
Thread.sleep(5000000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
执行结果为
file
newScheduledThreadPool
newScheduledThreadPool即为计划的线程池，源码为
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
可以看到jdk单独定义了一个具体的ScheduledThreadPoolExecutor类来完成这种定时任务的操作。可以看到最大线程数为Integer.MAX_VALUE，这样也可能因为线程数过多造成内存溢出。但是因为keepAliveTime为0，所以当线程执行完后立即停止运行。另外此处用的是一个优先队列，同样也是一个无界的阻塞队列。即也有可能因为队列任务过多造成OOM异常，另外也有可能因为最大线程数为Integer.MAX_VALUE，创建了太多的线程造成OOM异常。具体的造成OOM异常的代码可以参考上述几种线程池的代码进行尝试。有如下两种方法
- scheduledThreadPool.schedule(new Task(),1, TimeUnit.SECONDS); 只在当前时间后的delay时间执行一次
- scheduledThreadPool.scheduleAtFixedRate(new Task(),2,3,TimeUnit.SECONDS);在当前时间后的初始时间后，以间隔每period时间执行一次
newWorkStealingPool
newWorkStealingPool 是jdk1.8新增的一种线程池，newWorkStealingPool的源码为
public static ExecutorService newWorkStealingPool(int parallelism) {
return new ForkJoinPool
(parallelism,
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
}
底层并没有使用原生的ThreadPoolExecutor而是用的ForkJoinPool 来实现的。 ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，把一个任务拆分成多个“小任务”分发到不同的cpu核心上执行，执行完后再把结果收集到一起返回。
public class WorkStealingPool {
public static void main(String[] args) throws InterruptedException {
ExecutorService workStealingPool = Executors.newWorkStealingPool(2);
for (int i = 1; i < 10; i++) {
int finalI = i;
workStealingPool.execute(() -> {
try {
System.out.println(finalI);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
Thread.sleep(10000);//这里让主线程等待子线程执行完毕，也可以使用计数器的方式
workStealingPool.shutdown();
}
}
总结
上文主要介绍了几种线程池的用法，后续文章会继续探讨自定义线程池以及线程池的使用
欢迎关注和点赞，以及总结的分类面试题https://github.com/zhendiao/JavaInterview
网友评论