线程池原理及应用

2019-02-15 Java并发编程 JavaSE, Java并发编程评论

一、为什么要使用线程池

在一些需要使用线程去处理任务的业务场景中，如果每一个任务都创建一个线程去处理，任务处理完过后，把这个线程销毁，这样会产生大量的线程创建、销毁的资源开销，Java 中更是如此，虚拟机将试图跟踪每一个对象。以便可以在对象销毁后进行垃圾回收。所以提高服务程序效率的一个手段就是尽可能降低创建和销毁对象的次数。使用线程池能够有效的控制这种线程的创建和销毁，而且能够对创建的线程进行有效的管理。

使用线程池的好处

降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度。当任务到达时，任务可以不需要等待线程创建就能立即执行。
提高线程的可管理性。线程时稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会
=降低系统稳定性，使用线程池可以进行统一分配、调优和监控。

线程池原理-概念

线程池管理器:用于创建并管理线程池,包括创建线程池,销毁线程池,添加新任务;
工作线程:线程池中线程,在没有任务时处于等待状态,可以循环的执行任务;
任务接口:每个任务必须实现的接口,以供工作线程调度任务的执行,它主要规定了任务的入
口,任务执行后的首尾工作,任务的执行状态等; 任务接口通常就是 Runable 接口,所以提交任务时,只要提交一个 Runable 的实现类就可以了
任务队列:用于存放没有处理的任务.提供一种缓存机制.

二、Java 线程池相关 API 介绍

Executor 接口
主要是用来执行提交的任务。
线程池会实现这个接口，并且使用 exectue 方法来提交一个任务。
ExecutorSevice 接口
ExecutorService 接口是 Executor 接口的一个子接口，它在 Executor 接口的基础上增加了一些方法，用来支持对任务的终止管理以及对异步任务的支持。
AbstractExecutorService 抽象类
AbstractExecutorService 实现了 ExcutorService，并且基于模板方法模式对一些方法给出了实现。是后面提到的线程池类 ThreadPoolExcutor 的直接父类。
ThreadPoolExcutor 类
ThreadPoolExcutor 通常就是我们所说的线程池类，Java 的线程池就是用这个类进行创建的。
在分析线程池的运行原理时，也是基于这个类来进行分析。
ScheduledExecutorService 接口
ScheduledExecutorService 接口时 ExecutorService 子接口，定义了线程池基于任务调度的一些方法。
ScheduledThreadPoolExecutor 类
ScheduledThreadPoolExecutor 集成了 ThreadPoolExecutor 类，并且实现了 ScheduledExecutorService 接口，对任务调度的功能进行了实现。
Executors 类
Executors 可以认为是线程池的工厂类，里面提供了静态方法对线程池进行创建。

三、Java 线程池的运行原理

1. 线程池的参数属性介绍

核心线程数 corePoolSize：核心线程池数量。
提交一个任务的时候，会对线程池里面的当前存活线程数量和这个 corePoolSize 进行比较，不同的情况下会有不同的操作。
最大线程数 maximumPoolSize：线程池所能创建的线程的最大数量。
空闲线程的超时时间 keepAliveTime：如果线程池当前的线程数大于 corePoolSize，并且这些线程中是有空闲线程的，也就是说这些线程没有在执行任务，那么空闲时间超过 keepAliveTime 时间，这些线程也会被销毁，指代前线程代数等于 corePoolsize，这时即便有空闲线程并且超时了，也不会进行线程销毁。
任务队列 workQueue：这是一个阻塞队列，用于存储提交的任务。
线程工厂 threadFactory：线程池会使用这个工厂类来创建线程，用户可以自己实现。
任务的拒绝处理 handler（RejectedExeutionHandler）：在线程数已经达到了最大线程数，而且任务队列也满了以后，提交的任务会使用这个 handler 来处理，用户也可以自己实现。默认是抛出一个异常 RejectedExecutionException。

2. 线程池运行原理分析

分析当用户提交一个任务时，线程池内部使如何运行的。
线程池原理

创建一个线程池，在还没有任务提交的时候，默认线程池里面是没有线程的。当然，可以调用 prestartCoreThread 方法，来预先创建一个核心线程。
线程池里面还没有线程或者线程池里面存活的线程数小于核心线程数 corePoolSize 时，这时对于一个新提交的任务，线程池会创建一个线程去处理提交的任务。当线程池里面存活的线程数小于等于核心线程数 corePoolSize 时，线程池里面的线程会一直存活着，就算空闲时间超过了 keepAliveTime，线程也不会被销毁，而是一直阻塞在那里一直等待任务队列的任务来执行。
当线程池里面存活的线程数已经等于 corePoolSize 了，这时对于一个新提交的任务，会被放进任务队列 workQueue 排队等待执行。而之前创建的线程并不会被销毁，而是不断的去拿阻塞队列里面的任务，当任务列表为空时，线程会阻塞，直到有任务被放进任务队列，线程拿到任务后继续执行，执行完了以后继续去拿任务，这也是为什么线程池队列要使用阻塞队
列。
当线程池里面存活的线程数已经等于 corePoolSize 了，并且任务队列也满了，这里假设 maximumPoolSize>corePoolSize（如果等于的话，就直接拒绝了），这时如果再来新的任务，线程池就会继续创建新的线程来处理新任务，直到线程数达到 maximumPoolSize，就不会再创建了。这些新创建的线程执行完了当前任务后，再任务队列里面还有任务的时候也不会销毁，而是去任务队列拿任务出来执行。在当前线程数大于 corePoolSize 过后，线程执行完当前任务，会有一个判断当前线程是否需要销毁的逻辑；如果能从任务队列中拿到任务，那么继续执行，如果拿任务时阻塞（说明队列中没有任务），那超过 keepAliveTime 时间就直接返回 null 并且销毁当前线程，直到线程池里面的线程数等于 corePoolSize 之后才不会进行线程销毁。
如果当前线程数达到了 maximumPoolSize，并且任务队列也满了，这种情况下还有新的任务过来，那就直接采用拒绝的处理器进行执行，默认的处理器逻辑时抛出一个 RejectedExcutionException 异常。你也可以指定其他的处理器，或者自定义一个拒绝处理器来实现拒绝逻辑的处理（比如把任务存储起来）。JDK 提供了四种拒绝策略处理类；
AbortPolicy（抛出一个异常，默认的），
DiscardPolicy（直接丢弃任务），
DiscardOldestPolicy（丢弃队列里最老的任务，将当前这个任务继续提交给线程池），
CallerRunPolicy（交给线程池调用的所在线程进行处理）。

图解线程池

3. 线程池包含以下四个基本组成部分：

线程池管理器（ThreadPool）：用于创建并管理线程池。包含创建线程池，销毁线程池，
加入新任务；
工作线程（PoolWorker）：线程池中线程，在没有任务时处于等待状态。能够循环的运行
任务。
任务接口（Task）：每一个任务必须实现的接口，以供工作线程调度任务的运行。它主要
规定了任务的入口。任务运行完成后的收尾工作，任务的运行状态等。
任务队列（taskQueue）：用于存放没有处理的任务。提供一种缓冲机制。

4. 常用的几种线程池以及使用场景

SingleThreadExecutor:单个线程的线程池
这种线程池主要适用于请求量非常小的场景，或者离线的数据处理等，只需要一个线程就够了。在持续的请求量比较大的情况下，不要使用这种线程池，单线程处理会使队列
不断变大，最终可能导致内存溢出。
FixedThreadPool：固定线程大小线程池
这种线程的额 corePoolSize 和 maximumPoolSize 是相等的，keepAliveTime 设置为 0，队列用的是 LinkedBlockingQueue 无界队列。适用于流量比较稳定的情况，不会说一段时间突然有大量的流量涌入，导致 LinkedBlockingQueue 越来越大最后导致内存溢出。
CachedThreadPool：按需求创建线程数量线程池
这种线程的 corePoolSize=0，maximumPoolSize 是 Integer.MAX_VALUE，
keepAliveTime 为 60 秒，队列使用 SynchronousQueue 同步队列，这个队列可以理解为没有容量的阻塞队列，只有有别的线程来拿任务时，当前线程才能插入成功，反过来也一样。所以这种线程池任务队列时不存任务的，任务全靠创建新的线程来处理，处理完了以后线程空闲超过 60 秒就会被自动销毁，所以这种线程池适合有一定高峰流量的场景。但是还是要慎用，如果瞬时流量过高会导致创建的线程过多，直接导致服务所在机器的 CPU 负载过高，然后卡死，所以使用这种线程池必须指代最高峰时的流量也不会导致 CPU 负载过高。
ScheduledThreadPoolExecutor：任务调度线程池
可以根据自己的需求，使用单线程调度（SingleThreadExecutor）,多线程调度（ScheduledThreadPool）。不过现在使用 spring 调度比较多，所以开发中比较少用。
自定义线程池（推荐使用）
根据实际的一个业务场景，自己 new 一个 ThreadPoolExecutor，参数根据业务场景需要指定合适的参数，比如核心线程数设置多少合适，最大线程数设置多少合适，任务队列设置多大的有界合适，拒绝策略也可以自定义，一般采用离线存储啥的，完全根据业务场景来定制。这样可以保证不会发生无界队列导致内存溢出，也不会导致创建的线程过多而导致机器卡死。

使用自定义线程池创建出 Executors 中提供的四种线程池

参数列表:

corePoolSize - 线程池核心池的大小。
maximumPoolSize - 线程池的最大线程数。
keepAliveTime - 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。
unit - keepAliveTime 的时间单位。
workQueue - 用来储存等待执行任务的队列。
threadFactory - 线程工厂。
handler - 拒绝策略。

SingleThreadExecutor

1	new ThreadPoolExecutor(1, 1,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());

FixedThreadPool

1	new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

CachedThreadPool

1	new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());

SingleThreadExecutor
1
2

5. 线程池关闭

shutdown():调用后不允许提交新任务，所有调用之前提交的任务都会执行，等所有任务
执行完，才会真正关闭线程池。
shutdownNow():强制关闭。返回还没有执行的 task 列表，然后不让等待的 task 执行，尝试
停止正在执行的 task。

线程池的使用案例

/** 线程池的使用 */
public class Demo {

    /**
     * 测试： 提交15个执行时间需要3秒的任务,看线程池的状况
    *
     * @param threadPoolExecutor 传入不同的线程池，看不同的结果
     * @throws Exception
    */
    public void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
        // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 查看线程数量，查看队列等待数量
        Thread.sleep(500L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
        // 等待15秒，查看线程数量和队列数量（理论上，会被超出核心线程数量的线程自动销毁）
        Thread.sleep(15000L);
        System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
        System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
    }

    /**
    * 1、线程池信息： 核心线程数量5，最大数量10，无界队列，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest1() throws Exception {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>());
        testCommon(threadPoolExecutor);
        // 预计结果：线程池线程数量为：5,超出数量的任务，其他的进入队列中等待被执行
    }

    /**
    * 2、 线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest2() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.err.println("有任务被拒绝执行了");
                    }
                });
        testCommon(threadPoolExecutor);
        // 预计结果：
        // 1、 5个任务直接分配线程开始执行
        // 2、 3个任务进入等待队列
        // 3、 队列不够用，临时加开5个线程来执行任务(5秒没活干就销毁)
        // 4、 队列和线程池都满了，剩下2个任务，没资源了，被拒绝执行。
        // 5、 任务执行，5秒后，如果无任务可执行，销毁临时创建的5个线程
    }

    /**
    * 3、 线程池信息： 核心线程数量5，最大数量5，无界队列，超出核心线程数量的线程存活时间：5秒
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest3() throws Exception {
        // 和Executors.newFixedThreadPool(int nThreads)一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        testCommon(threadPoolExecutor);
        // 预计结：线程池线程数量为：5，超出数量的任务，其他的进入队列中等待被执行
    }

    /**
    * 4、 线程池信息：
    * 核心线程数量0，最大数量Integer.MAX_VALUE，SynchronousQueue队列，超出核心线程数量的线程存活时间：60秒
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest4() throws Exception {

        // SynchronousQueue，实际上它不是一个真正的队列，因为它不会为队列中元素维护存储空间。与其他队列不同的是，它维护一组线程，这些线程在等待着把元素加入或移出队列。
        // 在使用SynchronousQueue作为工作队列的前提下，客户端代码向线程池提交任务时，
        // 而线程池中又没有空闲的线程能够从SynchronousQueue队列实例中取一个任务，
        // 那么相应的offer方法调用就会失败（即任务没有被存入工作队列）。
        // 此时，ThreadPoolExecutor会新建一个新的工作者线程用于对这个入队列失败的任务进行处理（假设此时线程池的大小还未达到其最大线程池大小maximumPoolSize）。

        // 和Executors.newCachedThreadPool()一样的
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());
        testCommon(threadPoolExecutor);
        // 预计结果：
        // 1、 线程池线程数量为：15，超出数量的任务，其他的进入队列中等待被执行
        // 2、 所有任务执行结束，60秒后，如果无任务可执行，所有线程全部被销毁，池的大小恢复为0
        Thread.sleep(60000L);
        System.out.println("60秒后，再看线程池中的数量：" + threadPoolExecutor.getPoolSize());
    }

    /**
    * 5、 定时执行线程池信息：3秒后执行，一次性任务，到点就执行 <br/>
    * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest5() throws Exception {
        // 和Executors.newScheduledThreadPool()一样的
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        threadPoolExecutor.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("任务被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 3000, TimeUnit.MILLISECONDS);
        System.out.println(
                "定时任务，提交成功，时间是：" + System.currentTimeMillis() + ", 当前线程池中线程数量：" + threadPoolExecutor.getPoolSize());
        // 预计结果：任务在3秒后被执行一次
    }

    /**
    * 6、 定时执行线程池信息：线程固定数量5 ，<br/>
    * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest6() throws Exception {
        ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
        // 周期性执行某一个任务，线程池提供了两种调度方式，这里单独演示一下。测试场景一样。
        // 测试场景：提交的任务需要3秒才能执行完毕。看两种不同调度方式的区别
        // 效果1： 提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，完毕后立刻执行)。
        // 也就是说这个代码中是，3秒钟执行一次（计算方式：每次执行三秒，间隔时间1秒，执行结束后马上开始下一次执行，无需等待）
        threadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务-1 被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 2000, 1000, TimeUnit.MILLISECONDS);

        // 效果2：提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，等上一次执行完毕后再开始计时，等待1秒)。
        // 也就是说这个代码钟的效果看到的是：4秒执行一次。 （计算方式：每次执行3秒，间隔时间1秒，执行完以后再等待1秒，所以是 3+1）
        threadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务-2 被执行，现在时间：" + System.currentTimeMillis());
            }
        }, 2000, 1000, TimeUnit.MILLISECONDS);
    }

    /**
    * 7、 终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest7() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.err.println("有任务被拒绝执行了");
                    }
                });
        // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        threadPoolExecutor.shutdown();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdown后，不接收新的任务，等待13任务执行结束
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }

    /**
    * 8、 立刻终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
    *
    * @throws Exception
    */
    private void threadPoolExecutorTest8() throws Exception {
        // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
        // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        System.err.println("有任务被拒绝执行了");
                    }
                });
        // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
        for (int i = 0; i < 15; i++) {
            int n = i;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("开始执行：" + n);
                        Thread.sleep(3000L);
                        System.err.println("执行结束:" + n);
                    } catch (InterruptedException e) {
                        System.out.println("异常：" + e.getMessage());
                    }
                }
            });
            System.out.println("任务提交成功 :" + i);
        }
        // 1秒后终止线程池
        Thread.sleep(1000L);
        List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
        // 再次提交提示失败
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("追加一个任务");
            }
        });
        System.out.println("未结束的任务有：" + shutdownNow.size());

        // 结果分析
        // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
        // 2、调用shutdownnow后，队列中的3个线程不再执行，10个线程被终止
        // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
    }

    public static void main(String[] args) throws Exception {
//      new Demo().threadPoolExecutorTest1();
//      new Demo().threadPoolExecutorTest2();
//      new Demo().threadPoolExecutorTest3();
//      new Demo().threadPoolExecutorTest4();
//      new Demo().threadPoolExecutorTest5();
//      new Demo().threadPoolExecutorTest6();
//      new Demo().threadPoolExecutorTest7();
        new Demo().threadPoolExecutorTest8();
    }
}

本文链接： https://yimchengjie.github.io/2019/02/15/Java%E5%B9%B6%E5%8F%91%E7%BC%96%E7%A8%8B/%E7%BA%BF%E7%A8%8B%E6%B1%A0%E5%8E%9F%E7%90%86%E5%8F%8A%E5%BA%94%E7%94%A8/

版权声明： 本博客所有文章除特别声明外，均采用 CC BY 4.0 CN协议许可协议。转载请注明出处！

yanchengjielove and share

Anything can go right will go right