本章继续学习Executors工具类。
1.Executors接口概述
@since 1.5
这个类定义了供Executor、ExecutorService、ScheduledExecutorService、ThreadFactory和Callable这些接口和类使用的工厂方法和工具方法。
Executors来自java.util.concurrent,是Executor并发框架的主要工具类。
Executors提供了以下几类方法:
第1类静态方法:将Runnable转换成Callable。
第2类静态方法:线程工厂(ThreadFactory)类。
第3类静态方法:实例化几类不可配置的线程池(ExecutorService和ScheduleExecutorService)。
第4类静态方法:实例化几类预先配置的常用线程池(ExecutorService)。
第5类静态方法:实例化几类预先配置的常用可调度线程池(ScheduleExecutorService)。
本章主要对第4、5类静态方法进行说说明。
2.第4类静态方法:几类预先配置的常用线程池
Executors通过静态方法提供以下几类预先配置的常用线程池:
2.1.newSingleThreadExecutor - 单任务线程池
使用一个单独的工作线程和*工作队列的线程池。
需要注意的是:如果这个工作线程在关闭之前因为执行失败而终止,则如果需要去执行后续任务,可以新建一个线程代替它。
任务是按顺序执行的,任意时刻,都不会有超过一个以上的活动线程。
不同于等效的newFixedThreadPool(1),newSingleThreadExecutor不能通过配置而达到使用额外线程的目的。
方法定义:
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
从方法定义可知:
其实际是通过ThreadPoolExecutor构造的。
使用了代理类FinalizableDelegatedExecutorService,使之无法修改配置。
参数说明
corePoolSize = 1 --> 1.至多只有一个活动线程会长期存在
maximumPoolSize = 1 --> 3.因为*队列,此参数无实际意义
keepAliveTime = 0L --> 4.因为*队列,此参数无实际意义
TimeUnit = TimeUnit.MILLISECONDS --> 4.因为*队列,此参数无实际意义
workQueue = new LinkedBlockingQueue<Runnable>()) --> 2.*队列,如果无可用核心线程,则新任务在此等待,不会再创建线程
实例代码:
System.out.println("===================== newSingleThreadExecutor - 单任务线程池"); //定义一个单任务线程池 ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor(); //循环执行5个任务 for (int i = 0; i < 5; i++) { //永远都是thread-1 singleThreadExecutor.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } Thread.sleep(1000); singleThreadExecutor.shutdown(); System.out.println("===================== newSingleThreadExecutor - 单任务线程池");
运行结果:
===================== newSingleThreadExecutor - 单任务线程池 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 ===================== newSingleThreadExecutor - 单任务线程池
结果说明:
从运行结果可知,无论向线程池中提交多少任务,其内只是pool-1-thread-1这个线程在执行。
2.2.newCachedThreadPool - 缓存线程池
创建一个线程池,这个线程池能够按需创建新线程,并且能够重用的之前创建的可用线程。
这个线程池会典型的提高处理多个短期异步任务的程序的新能。
如果有可用的线程,执行任务会尽量重用以前构建的线程。
如果没有可用的线程,将会创建一个新的线程,并将此线程添加到线程池中。
空闲超过60秒的线程将会被终止,并且从缓存中移除。
因此,即使这个线程池空闲再长时间,也不会消耗任何资源。
注意,可以通过ThreadPoolExecutor的构造函数,构造具有相似属性不同细节(例如:超时参数)的缓存线程池实现。
方法定义:
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
从方法定义可知:
其实际是通过ThreadPoolExecutor构造的。
参数说明:
corePoolSize = 0 --> 1.不会存在长期存在的线程
maximumPoolSize = Integer.MAX_VALUE --> 3.最多创建2,147,483,647个线程
keepAliveTime = 60L --> 4.线程执行完任务后,最多空闲60秒,就会关闭
TimeUnit = TimeUnit.SECONDS --> 4.线程执行完任务后,最多空闲60秒,就会关闭
workQueue = new SynchronousQueue<Runnable>() --> 2.直传队列,新线程到达,不会等待,会直接创建新线程
实例代码:
System.out.println("===================== newCachedThreadPool - 缓存线程池"); //定义一个缓冲线程池 ExecutorService cachedThreadPool = Executors.newCachedThreadPool(); //创建5个线程 for (int i = 0; i < 5; i++) { //如果无可用线程,则创建新线程 cachedThreadPool.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } //等待1秒 Thread.sleep(1000); System.out.println(); //再次重新创建5个线程 for (int i = 0; i < 5; i++) { //如果有可用线程,则重用之前创建的线程 cachedThreadPool.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } //等待1秒 Thread.sleep(70000); System.out.println(); //再次重新创建5个线程 for (int i = 0; i < 5; i++) { //如果有可用线程,则重用之前创建的线程 cachedThreadPool.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } Thread.sleep(1000); cachedThreadPool.shutdown(); System.out.println("===================== newCachedThreadPool - 缓存线程池")
运行结果:
===================== newCachedThreadPool - 缓存线程池 pool-2-thread-1 pool-2-thread-3 pool-2-thread-4 pool-2-thread-5 pool-2-thread-2 pool-2-thread-3 pool-2-thread-4 pool-2-thread-1 pool-2-thread-2 pool-2-thread-5 pool-2-thread-6 pool-2-thread-7 pool-2-thread-8 pool-2-thread-9 pool-2-thread-10 ===================== newCachedThreadPool - 缓存线程池
结果说明:
当无可用的空闲线程时,线程池会创建新的线程来执行新的任务。
当有可用的空闲线程时,线程池会优先使用之前创建的线程来执行新的任务。
线程的空闲时间超过60秒,就会关闭;这时再进来新的任务,线程池又会创建新线程来执行任务。
2.3.newFixedThreadPool - 固定大小线程池
创建一个线程池,这个线程池重复使用固定数量的线程池,以及一个共享的*队列。
在任何时候,最多有nThreads个活动的线程。
如果所有的nThreads个线程都处于活动状态,则新提交的任务将会在队列中等待。
如果一个工作线程在关闭之前因为执行失败而终止,则如果需要去执行后续任务,可以新建一个线程代替它。
线程池中的线程会一直存在,直到显式的调用关闭方法shutdown或shutdownNow。
方法定义:
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
从方法定义可知:
其实际是通过ThreadPoolExecutor构造的。
参数说明:
corePoolSize = nThreads --> 1.至多有nThreads个活动线程会长期存在
maximumPoolSize = nThreads --> 3.因为*队列,此参数无实际意义
keepAliveTime = 0L --> 4.因为*队列,此参数无实际意义
TimeUnit = TimeUnit.MILLISECOND --> 4.因为*队列,此参数无实际意义
workQueue = new LinkedBlockedQueue<Runnable>()> --> 2.*队列,如果无可用核心线程,则新任务在此等待,不会再创建线程
实例代码:
System.out.println("===================== newFixedThreadPool - 固定大小线程池"); ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2); for (int i = 0; i < 5; i++) { //在thread-1 和 Thread-2 之间切换 fixedThreadPool.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } Thread.sleep(1000); fixedThreadPool.shutdown(); System.out.println("===================== newFixedThreadPool - 固定大小线程池");
运行结果:
从运行结果可知,无论向线程池中提交多少任务,其内只有2个线程在执行。
2.4.newWorkStealingPool - 并行工作者线程池
- 创建一个工作窃取线程池,以JVM运行时可以CPU核数作为线程池的并行度。
- 此线程池通过ForkJoinPool实现。
方法定义:
public static ExecutorService newWorkStealingPool() { return new ForkJoinPool (Runtime.getRuntime().availableProcessors(), ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true); }
参数说明:
parallelism = Runtime.getRuntime().availableProcessors() --> 1.并行级别为运行期可用的CPU处理器数量
factory = ForkJoinPool.defaultForkJoinWorkerThreadFactory --> 2.使用ForkJoinPool默认的工作线程队列
handler = null --> 3.不设置异常捕捉处理器
asyncMode = true --> 4.采取异步模式
实例代码:
System.out.println("===================== newWorkStealingPool - 并行任务线程池"); //定义一个并行工作者线程池 ExecutorService workStealingPool = Executors.newWorkStealingPool(); //循环创建线程池 for (int i = 0; i < 20; i++) { //因为本机CPU为4核,所以并行级别为4,即最多平行四个工作者。 workStealingPool.submit(() -> { System.out.println(Thread.currentThread().getName()); }); } // Thread.sleep(1000); workStealingPool.shutdown(); System.out.println("===================== newWorkStealingPool - 并行任务线程池");
运行结果:
===================== newWorkStealingPool - 并行任务线程池 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-2 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-2 ForkJoinPool-1-worker-2 ForkJoinPool-1-worker-2 ForkJoinPool-1-worker-2 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-3 ForkJoinPool-1-worker-0 ForkJoinPool-1-worker-1 ForkJoinPool-1-worker-2 ===================== newWorkStealingPool - 并行任务线程池
结果说明:
因为本机CPU为4核,所以并行级别为4,即最多平行四个工作者。
3.第5类静态方法:几类预先配置的常用可调度线程池
Executors通过静态方法提供以下几类预先配置的常用可调度线程池:
3.1.newScheduledThreadPool - 固定大小调度线程池
创建一个线程池,这个线程池可以延时执行任务,或者周期性的执行任务。
方法定义:
//Executors public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } //ScheduledThreadPoolExecutor public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); } //ThreadPoolExecutor public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
从方法定义可知:
其实际是通过ScheduledThreadPoolExecutor构造的.
而ScheduledThreadPoolExecutor是通过ThreadPoolExecutor构造的。
参数说明:
corePoolSize = 指定大小 --> 1.至多有指定数量的活动线程会长期存在
maximumPoolSize = Integer.MAX_VALUE --> 3.因为*队里,此参数无实际意义
keepAliveTime = 0L --> 4.因为*队列,此参数无实际意义
TimeUnit = TimeUnit.NANOSECONDS --> 4.因为*队列,此参数无实际意义
workQueue = new DelayedWorkQueue() --> 2.一种*队列,如果无可用核心线程,则新任务在此等待,不会再创建线程
实例代码:
System.out.println("===================== newScheduledThreadPool - 调度线程池"); //定义一个固定大小的调度线程池 ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2); //循环提交任务 for (int i = 0; i < 5; i++) { //在thread-1 和 Thread-2 之间切换 //如果不能计算好线程池的核心线程数量和任务延时之间的关系,很可能造成指定的延时任务并未按照计划执行 scheduledThreadPool.schedule(() -> { System.out.println(Thread.currentThread().getName() + " begin... "); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " end. "); }, 2, TimeUnit.SECONDS); } Thread.sleep(7000); System.out.println("===================== newScheduledThreadPool - 调度线程池"); scheduledThreadPool.shutdown();
运行结果:
===================== newScheduledThreadPool - 调度线程池 pool-4-thread-2 begin... pool-4-thread-1 begin... pool-4-thread-2 end. pool-4-thread-2 begin... pool-4-thread-1 end. pool-4-thread-1 begin... pool-4-thread-2 end. pool-4-thread-2 begin... pool-4-thread-1 end. pool-4-thread-2 end. ===================== newScheduledThreadPool - 调度线程池
结果说明:
这些任务都在2秒之后才开始执行。
这些任务只会通过pool-4-thread-1和pool-4-thread-2这两个工作线程执行。
3.2.newSingleThreadScheduledExecutor - 单线程的调度线程池
创建一个单线程线程池,这个线程池可以延时执行任务,或者周期性的执行任务。
但是请注意,如果这个工作线程在关闭之前因为执行失败而终止,则如果需要去执行后续任务,可以新建一个线程代替它。
任务是按顺序执行的,任意时刻,都不会有超过一个以上的活动线程。
不同于等效的newScheduledThreadPool(1),newSingleThreadScheduledExecutor不能通过配置而达到使用额外线程的目的。
方法定义:
//Executors public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new DelegatedScheduledExecutorService (new ScheduledThreadPoolExecutor(1)); } //Executors DelegatedScheduledExecutorService(ScheduledExecutorService executor) { super(executor); e = executor; } //Executors DelegatedExecutorService(ExecutorService executor) { e = executor; } //ScheduledThreadPoolExecutor public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); } //ThreadPoolExecutor public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
从方法定义可知:
使用了代理类DelegatedScheduledExecutorService,使之无法修改配置。
其实际是通过ScheduledThreadPoolExecutor构造的.
而ScheduledThreadPoolExecutor是通过ThreadPoolExecutor构造的。
参数说明:
corePoolSize = 1 --> 1.至多只有1个的活动线程会长期存在
maximumPoolSize = Integer.MAX_VALUE --> 3.因为*队里,此参数无实际意义
keepAliveTime = 0L --> 4.因为*队列,此参数无实际意义
TimeUnit = TimeUnit.NANOSECONDS --> 4.因为*队列,此参数无实际意义
workQueue = new DelayedWorkQueue() --> 2.一种*队列,如果无可用核心线程,则新任务在此等待,不会再创建线程
实例代码:
System.out.println("===================== newSingleThreadScheduledExecutor - 单线程的调度线程池"); //定义一个单线程的调度线程池 ScheduledExecutorService singleThreadScheduledExecutor = Executors.newSingleThreadScheduledExecutor(); for (int i = 0; i < 5; i++) { singleThreadScheduledExecutor.schedule(() -> { System.out.println(Thread.currentThread().getName() + " begin... "); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " end. "); }, 2, TimeUnit.SECONDS); } Thread.sleep(10000); System.out.println("===================== newSingleThreadScheduledExecutor - 单线程的调度线程池"); singleThreadScheduledExecutor.shutdown();
运行结果:
===================== newSingleThreadScheduledExecutor - 单线程的调度线程池 pool-5-thread-1 begin... pool-5-thread-1 end. pool-5-thread-1 begin... pool-5-thread-1 end. pool-5-thread-1 begin... pool-5-thread-1 end. pool-5-thread-1 begin... pool-5-thread-1 end. pool-5-thread-1 begin... pool-5-thread-1 end. ===================== newSingleThreadScheduledExecutor - 单线程的调度线程池
结果说明:
- 这些任务都在2秒之后才开始执行。
- 这些任务只会通过pool-4-thread-1这1个工作线程执行。