- 线程池介绍
- 并发队列
- 线程池原理分析
- 自定义线程池
文中部分代码使用 lambda 表达式以简化代码。
线程池
什么是线程池?
Java中的线程池是运用场景最多的并发框架,几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中,合理地使用线程池能够带来3个好处。
- 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提高响应速度。当任务到达时,任务可以不需要等到线程创建就能立即执行。
- 提高线程的可管理性。线程是稀缺资源,如果无限制地创建,不仅会消耗系统资源,还会降低系统的稳定性,使用线程池可以进行统一分配、调优和监控
线程状态
线程生命周期
新建状态
当用new操作符创建一个线程时, 例如new Thread(r),线程还没有开始运行,此时线程处在新建状态。
new Thread(() -> System.out.println("run 任务执行"))
就绪状态
一个新创建的线程并不自动开始运行,要执行线程,必须调用线程的start()方法。处于就绪状态的线程并不一定立即运行run()方法,线程还必须同其他线程竞争CPU时间,只有获得CPU时间才可以运行线程。
new Thread(() -> System.out.println("run 任务执行")).start();
运行状态
当线程获得CPU时间后,它才进入运行状态,真正开始执行run()方法。也就是 System.out.println("run 任务执行") 。
run 任务执行
阻塞状态
线程运行过程中,可能由于各种原因进入阻塞状态 :
-
线程执行了Thread.sleep(int n)方法,线程放弃CPU,睡眠n毫秒,然后恢复运行。
-
线程要执行一段同步代码,由于无法获得相关的同步锁,只好进入阻塞状态,等到获得了同步锁,才能恢复运行。
-
线程执行了一个对象的wait()方法,进入阻塞状态,只有等到其他线程执行了该对象的notify()或notifyAll()方法,才可能将其唤醒。
死亡状态
有两个原因会导致线程死亡:
- run方法正常退出而自然死亡
- 一个未捕获的异常终止了run方法而使线程死亡
使用 isAlive() 方法可判断线程在当前是否存活着
并发队列
在并发队列上JDK提供了两套实现,一个是以ConcurrentLinkedQueue为代表的高性能队列非阻塞队列,一个是以BlockingQueue接口为代表的阻塞队列,无论哪种都继承自Queue。
ConcurrentLinkedDeque 非阻塞式队列
- 无界线程安全队列
- 先进先出的原则
- 不允许null元素
重要方法:
- add() 和 offer() 都是加入元素的方法
- poll() 和 peek() 都是取头元素节点,区别在于前者会删除元素,后者不会
适用于高并发场景下的队列,通过无锁的方式,实现了高并发状态下的高性能,通常ConcurrentLinkedQueue性能好于BlockingQueue.它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。头是最先
加入的,尾是最近加入的,该队列不允许null元素public static void main(String[] args) { // 非阻塞式用法 无界队列 - 先进先出的原则 ConcurrentLinkedQueue concurrentLinkedQueue = new ConcurrentLinkedQueue(); concurrentLinkedQueue.offer("第一个进队列"); concurrentLinkedQueue.offer("第二个进队列"); // 获取单个队列信息 peek 获取队列 System.out.println(concurrentLinkedQueue.peek()); System.out.println(concurrentLinkedQueue.size()); // 获取单个队列信息 poll 获取队列之后 会删除队列信息 移除 System.out.println(concurrentLinkedQueue.poll()); System.out.println(concurrentLinkedQueue.size()); }
第一个进队列2第一个进队列1
BlockingQueue 阻塞式队列
被阻塞的情况主要有如下两种:
- 当队列满了的时候进行入队列操作
- 当队列空了的时候进行出队列操作
以下简单介绍 BlockingQueue 成员
ArrayBlockingQueue
有边界的阻塞队列,它的内部实现是一个数组,先进先出原则。使用阻塞必须加加超时时间
public static void main(String[] args) throws InterruptedException { // 阻塞式队列 ArrayBlockingQueue
new ArrayBlockingQueue<>(3) 设定(队列)有界大小3。
offer("张三") - “张三“ 进入队列 第一个 poll("张三") 取出数据,“张三”从队列中移除 第二个 poll("张三") ,此时队列已经没有数据了会阻塞等待3秒,如果队列在这3秒内有数据入队列会立即取出数据并结束阻塞,3秒阻塞超时就返回null张三null0
以上示例是 当队列空了的时候进行出队列阻塞
public static void main(String[] args) throws InterruptedException { // 阻塞式队列 ArrayBlockingQueue
new ArrayBlockingQueue<>(3) 设定(队列)有界大小3。
offer("张一") - “张一“ 进入队列 offer("张二") - “张二“ 进入队列 offer("张三") - “张三“ 进入队列 , 此时队列已满 offer("张四",3, TimeUnit.SECOND) 此时队列已满 , 这里给了3秒阻塞时间等待数据出列才有位置,期间如果有数据出列立即添加“张四”入队列并结束阻塞,如果没数据出列阻塞超时会废弃当前数据“张四”张一2
以上示例是 当队列满了的时候进行入队列阻塞
常见的阻塞式队列如下:
- LinkedBlockingQueue
- PriorityBlockingQueue
- SynchronousQueue
有兴趣可以去实践下
ThreadPoolExecutor
- Executor - execute 方法接口
- Executors - 工厂类
- ThreadPoolExecutor - Executor 框架的最顶层实现类
Executor框架的最顶层实现是ThreadPoolExecutor类,Executors工厂类中提供的newScheduledThreadPool、newFixedThreadPool、newCachedThreadPool 静态方法其实也只是ThreadPoolExecutor的构造函数参数不同而已。通过传入不同的参数,就可以构造出适用于不同应用场景下的线程池,来看下构造的参数。
public ThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueworkQueue) { ... }
- corePoolSize 线程池核心线程数,默认情况下,核心线程会在线程池中一直存活,即使它们处于闲置状态。如果ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true,那么闲置的核心线程在等待新任务到来时会有超时策略,超过keepAliveTime 时间后,核心线程就会被终止。
- maximumPoolSize 线程池所能容纳的最大线程数,当活动线程达到这个数值后,后续的新任务将被阻塞。
- keepAliveTime 非核心线程超过这个时长就会被回收。当ThreadPoolExecutor的allowCoreThreadTimeOut属性设置为true时,keepAliveTime同样会作用于核心线程
- unit keepAliveTime 参数的时间单位
- workQueue 执行前用于保持任务的队列, 此队列仅由 execute 方法提交的 Runnable 任务。
newCachedThreadPool
这里用 可缓存的 newCachedThreadPool 线程池来分析
ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor( 0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue()); }
newCachedThreadPool的corePoolSize被设置为0, maximumPoolSize被设置为Integer.MAX_VALUE, 即maximum是无界的。这里keepAliveTime设置为60秒,意味着空闲的线程最多可以等待任务60秒,否则将被回收。
newCachedThreadPool使用没有容量的SynchronousQueue作为主线程池的工作队列,它是一个不存储元素阻塞队列。
线程池中的线程是被线程池缓存了的,也就是说,线程没有任务要执行时,便处于空闲状态,处于空闲状态的线程并不会被立即销毁(会被缓存住),只有当空闲时间超出一段时间(默认为60s)后,线程池才会销毁该线程(相当于清除过时的缓存)。新任务到达后,线程池首先会让被缓存住的线程(空闲状态)去执行任务,如果没有可用线程(无空闲线程),便会创建新的线程。
根据特性可知 :
- 长时间不提交任务的CachedThreadPool不会占用系统资源
- 极端情况下,CachedThreadPool会因为创建过多线程而耗尽CPU资源及内存
代码示例
public static void main(String[] args) { // 创建线程(四种方式) 1.可缓存、定长、定时、单例 ExecutorService newCachedThreadPool = Executors.newCachedThreadPool(); for (int i = 0; i < 100; i++) { final int temp = i; // execute方法作用: 执行任务 newCachedThreadPool.execute(() -> { System.out.println(Thread.currentThread().getName() + ",i:" + temp); }); } }
newCachedThreadPool 线程池执行 100 次任务
pool-1-thread-3,i:2pool-1-thread-1,i:0pool-1-thread-2,i:1pool-1-thread-4,i:3pool-1-thread-5,i:4pool-1-thread-7,i:6pool-1-thread-6,i:5pool-1-thread-9,i:8pool-1-thread-10,i:9pool-1-thread-8,i:7pool-1-thread-11,i:10pool-1-thread-1,i:21pool-1-thread-3,i:20
可以看到线程 pool-1-thread-3、pool-1-thread-1 被重复使用了
newFixedThreadPool
定长的 newFixedThreadPool 线程池
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(3);
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor( nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue());}
可以看到 newFixedThreadPool 的核心线程和最大的线程都是一样的,最多3个线程将处于活动状态。如果提交了3个以上的线程,那么它们将保持在队列中,直到线程可用。
代码示例
public static void main(String[] args) { // 可固定长度的线程池 核心线程数 为3 最多创建3个線程 ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(3); for (int i = 0; i < 10; i++) { final int temp = i; // execute方法作用: 执行任务 newFixedThreadPool.execute(() -> { System.out.println(Thread.currentThread().getName() + ",i:" + temp); }); } }
pool-1-thread-3,i:2pool-1-thread-2,i:1pool-1-thread-1,i:0pool-1-thread-3,i:3pool-1-thread-2,i:5pool-1-thread-3,i:6pool-1-thread-1,i:4pool-1-thread-3,i:8pool-1-thread-2,i:7pool-1-thread-1,i:9
可以看到始终只有pool-1-thread-1、pool-1-thread-2、pool-1-thread-3 三个线程
newScheduledThreadPool
可定时的 newScheduledThreadPool 线程池
public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); }
newScheduledThreadPool 的核心线程为 传入的corePoolSize,最大线程为Integer.MAX_VALUE , DelayedWorkQueue队列实现BlockingQueue接口,所以使用阻塞式的BlockingQueue队列。
代码示例
public static void main(String[] args) { // 可定时线程池 3核心线程数 ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(3); newScheduledThreadPool.scheduleAtFixedRate( // Runnable 任务 () -> System.out.println(Thread.currentThread().getName() + "run "), 1000, // 初次执行需等待时间 100, TimeUnit.MILLISECONDS); // 周期执行时间(3个线程抢cpu时间) }
pool-1-thread-1run pool-1-thread-2run pool-1-thread-3run pool-1-thread-3run pool-1-thread-2run
每 100 毫秒执行一次
newSingleThreadExecutor
单例的 newSingleThreadExecutor 线程池
代码示例
public static void main(String[] args) { // 单线线程 ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor(); for (int i = 0; i < 10; i++) { final int temp = i; // execute方法作用: 执行任务 newSingleThreadExecutor.execute(() -> System.out.println(Thread.currentThread().getName() + ",i:" + temp)); } }
pool-1-thread-1,i:0pool-1-thread-1,i:1pool-1-thread-1,i:2pool-1-thread-1,i:3pool-1-thread-1,i:4pool-1-thread-1,i:5pool-1-thread-1,i:6pool-1-thread-1,i:7pool-1-thread-1,i:8pool-1-thread-1,i:9
始终只有一个线程执行任务
线程池原理剖析
提交一个任务到线程池中,线程池的处理流程如下:
-
1 .判断线程池里的核心线程是否都在执行任务,如果不是(核心线程空闲或者还有核心线程没有被创建)则创建一个新的工作线程来执行任务。如果核心线程都在执行任务,则进入下个流程。
-
2 .线程池判断工作队列是否已满,如果工作队列没有满,则将新提交的任务存储在这个工作队列里。如果工作队列满了,则进入下个流程。
-
3 . 判断线程池里的线程是否都处于工作状态,如果没有,则创建一个新的工作线程来执行任务。如果已经满了,则交给饱和策略来处理这个任务。
-
给大家推荐一个交流学习qq群:698581634 进群即可免费获取资料。
自定义线程线程池
上面列举到 newCachedThreadPool、newFixedThreadPool ... 底层都是对ThreadPoolExecutor的构造器进行包装使用。在来看下
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); }
所以咱们如果要实现一个自定义线程池就 直接 newThreadPoolExecutor(...) 就就行,现在来自定义一个。
ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(4));
这里自定义线程池 :核心线程1个,最大创建2个线程,60s等待超时销毁,使用阻塞式有界ArrayBlockingQueue 队列,最多缓存4个任务。
public class CustomThread { public static void main(String[] args) { ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(4)); for (int i = 1; i <= 6; i++) { TaskThred t1 = new TaskThred("任务" + i); executor.execute(t1); } executor.shutdown(); }}class TaskThred implements Runnable { private String taskName; public TaskThred(String taskName) { this.taskName = taskName; } @Override public void run() { System.out.println(Thread.currentThread().getName() + taskName); }}
execute 执行6个任务
pool-1-thread-1任务1pool-1-thread-2任务6pool-1-thread-2任务2pool-1-thread-1任务4pool-1-thread-1任务5pool-1-thread-2任务3
相关的异常信息
RejectedExecutionException
原因:- 提交任务量大, 队列缓存较小
- 确保不要在shutdown()之后在执行任务
java.util.concurrent.RejectedExecutionException: Task com.snail.demo.TaskThred@74a14482 rejected from java.util.concurrent.ThreadPoolExecutor@1540e19d[Running, pool size = 2, active threads = 2, queued tasks = 2, completed tasks = 0] at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
要出现这个异常只需将上边 new ArrayBlockingQueue<>(4) 大小 4 改为 2,如下
ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 2, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(2));
原文链接:https://www.jianshu.com/p/3a3b00bb6ab9?hmsr=toutiao.io&utm_medium=toutiao.io&utm_source=toutiao.io