ICode9
精准搜索请尝试： 精确搜索
首页
编程语言>
数据库
系统相关
互联网
其他分享
Python
Java
JavaScript
android
PHP
首页 > 编程语言> 文章详细

Java并发42:Executor系列--Executors(下)-几类预定义的线程池ExecutorService和可调度线程池ScheduledExecutorService

2021-10-09 10:34:29  阅读：190  来源： 互联网

标签：thread Executors -- ForkJoinPool 线程 println pool


本章继续学习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个工作线程执行。

标签：thread,Executors,--,ForkJoinPool,线程,println,pool
来源： https://www.cnblogs.com/yaochunhui/p/15384422.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

关于我们 | 联系我们 | 留言反馈

专注分享技术，共同学习，共同进步。侵权联系[81616952@qq.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有