多线程编程学习十一(ThreadPoolExecutor 详解).

时间  2019-11-06
标签 多线程 编程 学习 十一 threadpoolexecutor 详解 栏目 Java 繁體版
原文   原文链接

1、ThreadPoolExecutor 参数说明

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
  • corePoolSize：核心线程池的大小。当提交一个任务到线程池时，核心线程池会建立一个核心线程来执行任务，即便其余核心线程可以执行新任务也会建立线程，等到须要执行的任务数大于核心线程池基本大小时就再也不建立。若是调用了线程池的 prestartAllCoreThreads() 方法，核心线程池会提早建立并启动全部核心线程。
  • workQueue：任务队列。当核心线程池中没有线程时，所提交的任务会被暂存在队列中。Java 提供了多种阻塞队列。
  • maximumPoolSize：线程池容许建立的最大线程数。若是队列也满了，而且已建立的线程数小于最大线程数，则线程池会再建立新的空闲线程执行任务。值得注意的是，若是使用了无界的任务队列则这个参数不起做用。
  • keepAliveTime：当线程池中的线程数大于 corePoolSize 时，keepAliveTime 为多余的空闲线程等待新任务的最长时间，超过这个时间后多余的线程将被终止。因此，若是任务不少，而且每一个任务执行的时间比较短，能够调大时间，提升线程的利用率。值得注意的是，若是使用了无界的任务队列则这个参数不起做用。
  • TimeUnit：线程活动保持时间的单位。

  • threadFactory：建立线程的工厂。能够经过线程工厂给每一个建立出来的线程设置符合业务的名字。html

    // 依赖 guava
new ThreadFactoryBuilder().setNameFormat("xx-task-%d").build();

  • handler：饱和策略。当队列和线程池都满了，说明线程池处于饱和状态，那么必须采起一种策略处理提交的新任务。Java 提供了如下4种策略：java

    • AbortPolicy：默认。直接抛出异常。
    • CallerRunsPolicy：只用调用者所在线程来运行任务。
    • DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
    • DiscardPolicy：不处理，丢弃掉。

tips： 通常咱们称核心线程池中的线程为核心线程，这部分线程不会被回收；超过任务队列后，建立的线程为空闲线程，这部分线程会被回收（回收时间即 keepAliveTime）数据库

2、常见的 ThreadPoolExecutor 介绍

Executors 是建立 ThreadPoolExecutor 和 ScheduledThreadPoolExecutor 的工厂类。小程序

Java 提供了多种类型的 ThreadPoolExecutor，比较常见的有 FixedThreadPool、SingleThreadExecutor、CachedThreadPool等。服务器

FixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

FixedThreadPool 被称为可重用固定线程数的线程池。能够看到 corePoolSize 和 maximumPoolSize 都被设置成了 nThreads；keepAliveTime设置为0L，意味着多余的空闲线程会被当即终止；使用了阻塞队列 LinkedBlockingQueue 做为线程的工做队列（队列的容量为 Integer.MAX_VALUE）。多线程

FixedThreadPool 所存在的问题是，因为队列的容量为 Integer.MAX_VALUE，基本能够认为是无界的，因此 maximumPoolSize 和 keepAliveTime 参数都不会生效，饱和拒绝策略也不会执行，会形成任务大量堆积在阻塞队列中。异步

FixedThreadPool 适用于为了知足资源管理的需求，而须要限制线程数量的应用场景。
函数


SingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

SingleThreadExecutor 是使用单个线程的线程池。能够看到 corePoolSize 和 maximumPoolSize 被设置为1，其余参数与 FixedThreadPool 相同，因此所带来的风险也和 FixedThreadPool 一致，就不赘述了。ui

SingleThreadExecutor 适用于须要保证顺序的执行各个任务。
线程


CachedThreadPool

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

CachedThreadPool 是一个会根据须要建立新线程的线程池。能够看到 corePoolSize 被设置为 0，因此建立的线程都为空闲线程；maximumPoolSize 被设置为 Integer.MAX_VALUE（基本可认为无界），意味着能够建立无限数量的空闲线程；keepAliveTime 设置为60L，意味着空闲线程等待新任务的最长时间为60秒；使用没有容量的 SynchronousQueue 做为线程池的工做队列。

CachedThreadPool 所存在的问题是， 若是主线程提交任务的速度高于maximumPool 中线程处理任务的速度时，CachedThreadPool 会不断建立新线程。极端状况下，CachedThreadPool会由于建立过多线程而耗尽CPU和内存资源。

CachedThreadPool 适用于执行不少的短时间异步任务的小程序，或者是负载较轻的服务器。

3、自建 ThreadPoolExecutor 线程池

鉴于上面提到的风险，咱们更提倡使用 ThreadPoolExecutor 去建立线程池，而不用 Executors 工厂去建立。

如下是一个 ThreadPoolExecutor 建立线程池的 Demo 实例：

public class Pool {

    static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("pool-task-%d").build();
    static ExecutorService executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors() * 2,
            200, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024),
            threadFactory, new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 无返回值的任务执行 -> Runnable
        executor.execute(() -> System.out.println("Hello World"));
        // 2. 有返回值的任务执行 -> Callable
        Future<String> future = executor.submit(() -> "Hello World");
        // get 方法会阻塞线程执行等待返回结果
        String result = future.get();
        System.out.println(result);

        // 3. 监控线程池
        monitor();

        // 4. 关闭线程池
        shutdownAndAwaitTermination();

        monitor();
    }

    private static void monitor() {
        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) Pool.executor;
        System.out.println("【线程池任务】线程池中曾经建立过的最大线程数：" + threadPoolExecutor.getLargestPoolSize());
        System.out.println("【线程池任务】线程池中线程数：" + threadPoolExecutor.getPoolSize());
        System.out.println("【线程池任务】线程池中活动的线程数：" + threadPoolExecutor.getActiveCount());
        System.out.println("【线程池任务】队列中等待执行的任务数：" + threadPoolExecutor.getQueue().size());
        System.out.println("【线程池任务】线程池已执行完任务数：" + threadPoolExecutor.getCompletedTaskCount());
    }

    /**
     * 关闭线程池
     * 1. shutdown、shutdownNow 的原理都是遍历线程池中的工做线程，而后逐个调用线程的 interrupt 方法来中断线程。
     * 2. shutdownNow：将线程池的状态设置成 STOP，而后尝试中止全部的正在执行或暂停任务的线程，并返回等待执行任务的列表。
     * 3. shutdown：将线程池的状态设置成 SHUTDOWN 状态，而后中断全部没有正在执行任务的线程。
     */
    private static void shutdownAndAwaitTermination() {
        // 禁止提交新任务
        executor.shutdown();
        try {
            // 等待现有任务终止
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                // 取消当前正在执行的任务
                executor.shutdownNow();
                // 等待一段时间让任务响应被取消
                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("Pool did not terminate");
                }
            }
        } catch (InterruptedException ie) {
            // 若是当前线程也中断，则取消
            executor.shutdownNow();
            // 保留中断状态
            Thread.currentThread().interrupt();
        }
    }
}

建立线程池须要注意如下几点：

  1. CPU 密集型任务应配置尽量小的线程，如配置 Ncpu+1 个线程。
  2. IO 密集型任务（数据库读写等）应配置尽量多的线程，如配置 Ncpu*2 个线程。
  3. 优先级不一样的任务可使用优先级队列 PriorityBlockingQueue 来处理。
  4. 建议使用有界队列。能够避免建立数量很是多的线程，甚至拖垮系统。有界队列能增长系统的稳定性和预警能力，能够根据须要设大一点儿，好比几千。

4、ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor 继承自 ThreadPoolExecutor。它主要用来在给定的延迟以后运行任务，或者按期执行任务。

public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory);
    }

ScheduledThreadPoolExecutor 的功能与 Timer 相似，但功能更强大、更灵活。Timer 对应的是单个后台线程，而ScheduledThreadPoolExecutor 能够在构造函数中指定多个对应的后台线程数。

Java 提供了多种类型的 ScheduledThreadPoolExecutor ，能够经过 Executors 建立，比较常见的有 ScheduledThreadPool、SingleThreadScheduledExecutor 等。适用于须要多个后台线程执行周期任务，同时为了知足资源管理的需求而须要限制后台线程数量的应用场景。

public class ScheduleTaskTest {

    static ThreadFactory threadFactory = new BasicThreadFactory.Builder().namingPattern("schedule-pool-%d").build();
    static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5, threadFactory);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 延迟 3 秒后执行 Runnable 方法
        scheduledExecutorService.schedule(() -> System.out.println("Hello World"), 3000, TimeUnit.MILLISECONDS);

        // 2. 延迟 3 秒后执行 Callable 方法
        ScheduledFuture<String> scheduledFuture = scheduledExecutorService.schedule(() -> "Hello ScheduledFuture", 3000, TimeUnit.MILLISECONDS);
        System.out.println(scheduledFuture.get());

        // 3. 延迟 1 秒后开始每隔 3 秒周期执行。
        //    若是中间任务遇到异常，则禁止后续执行。
        //    固定的频率来执行某项任务，它不受任务执行时间的影响。到时间，就执行。
        scheduledExecutorService.scheduleAtFixedRate(() -> System.out.println("Hello ScheduleAtFixedRate"), 1, 3000, TimeUnit.MILLISECONDS);

        // 4. 延迟 1 秒后，每一个任务结束延迟 3 秒后再执行下个任务。
        //    若是中间任务遇到异常，则禁止后续执行。
        //    受任务执行时间的影响，等待任务执行结束后才开始计算延迟。
        scheduledExecutorService.scheduleWithFixedDelay(() -> System.out.println("Hello ScheduleWithFixedDelay"), 1, 3000, TimeUnit.MILLISECONDS);
    }
}

ScheduledThreadPoolExecutor 的执行步骤大抵以下：

  1. 当调用 ScheduledThreadPoolExecutor 的 scheduleAtFixedRate() 方法或者 scheduleWithFixedDelay()方 法时，会向 DelayedWorkQueue 队列添加 ScheduledFutureTask 任务。
  2. 线程池中的线程从 DelayedWorkQueue队列中获取执行时间已到达的 ScheduledFutureTask，而后执行任务。
  3. 线程修改 ScheduledFutureTask 任务的执行时间为下次将要被执行的时间。
  4. 线程把修改后的 ScheduledFutureTask 从新放回队列。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程