阻塞队列&线程池

前言

阻塞队列是线程池的基础。两者都是面试热点，尤其是线程池。所以我特地花时间学习了一下这方面的知识，并做记录。

一.阻塞队列

这个其实用的非常多。安卓里面 ​​Handler​​​机制的​​MessageQueue​​，我认为就是一个阻塞队列。阻塞队列与一般的队列有何不同呢？我个人理解是：阻塞队列可以让访问它的线程阻塞掉。

1.阻塞队列的定义

1）支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满。
2）支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。

2.应用场景

我们知道的生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题的。生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力，提高了程序整体处理数据的速度。
生产者消费者模式是阻塞队列最普遍的用途。

3.阻塞队列常用方法以及解释

以上是阻塞队列里面的常用方法。最上面一行的后面四列意思是当队列满或者队列空的特殊情况的时候，这些方法分别会做出什么反应。比如当队列满的时候，再使用​​add​​​则会抛出异常，再使用​​offer​​​会返回特殊值，再使用​​put​​就会阻塞。

• 解释：
  抛出异常：当队列满时，如果再往队列里插入元素，会抛出​​IllegalStateException（"Queuefull"）​​​异常。当队列空时，从队列里获取元素会抛出​​NoSuchElementException​​​异常。
  ·返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回​​true​​​。如果是移除方法，则是从队列里取出一个元素，如果没有则返回​​null​​​。
  一直阻塞：当阻塞队列满时，如果生产者线程往队列里​​put​​​元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列里​​take​​​元素，队列会阻塞住消费者线程，直到队列不为空。
  ·超时退出：当阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超过了指定的时间，生产者线程就会退出。

4.常用的阻塞队列及其区别

以上的阻塞队列都实现了​​BlockingQueue​​接口，也都是线程安全的。

下面一一进行解释。

• ①​​ArrayBlockingQueue​​ 是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证线程公平的访问队列，所谓公平访问队列是指阻塞的线程，可以按照阻塞的先后顺序访问队列，即先阻塞线程先访问队列。非公平性是对先等待的线程是非公平的，当队列可用时，阻塞的线程都可以争夺访问队列的资格，有可能先阻塞的线程最后才访问队列。初始化时有参数可以设置
  ②​​LinkedBlockingQueue​​ 是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为​​Integer.MAX_VALUE​​。此队列按照先进先出的原则对元素进行排序。
  ③​​PriorityBlockingQueue​​​​PriorityBlockingQueue​​是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。也可以自定义类实现​​compareTo​​()方法来指定元素排序规则，或者初始化​​PriorityBlockingQueue​​时，指定构造参数​​Comparator​​来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。
  ④​​DelayQueue​​ 是一个支持延时获取元素的无界阻塞队列。队列使用​​PriorityQueue​​来实现。队列中的元素必须实现​​Delayed​​接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。
  ​​DelayQueue​​非常有用，可以将​​DelayQueue​​运用在以下应用场景。
  缓存系统的设计：可以用​​DelayQueue​​保存缓存元素的有效期，使用一个线程循环查询​​DelayQueue​​，一旦能从​​DelayQueue​​中获取元素时，表示缓存有效期到了。
  ⑤​​SynchronousQueue​​ 是一个不存储元素的阻塞队列。每一个​​put​​操作必须等待一个​​take​​操作，否则不能继续添加元素。​​SynchronousQueue​​可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。​​SynchronousQueue​​的吞吐量高于​​LinkedBlockingQueue​​和​​ArrayBlockingQueue​​。
  ⑥​​LinkedTransferQueue​​ 多了​​tryTransfer​​和​​transfer​​方法，
  （1）​​transfer​​方法
  如果当前有消费者正在等待接收元素（消费者使用​​take​​()方法或带时间限制的​​poll​​()方法时），​​transfer​​方法可以把生产者传入的元素立刻​​transfer​​（传输）给消费者。如果没有消费者在等待接收元素，​​transfer​​方法会将元素存放在队列的​​tail​​节点，并等到该元素被消费者消费了才返回。
  （2）​​tryTransfer​​方法
  ​​tryTransfer​​方法是用来试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回​​false​​。和​​transfer​​方法的区别是​​tryTransfer​​方法无论消费者是否接收，方法立即返回，而​​transfer​​方法是必须等到消费者消费了才返回。
  ⑦​​LinkedBlockingDeque​​​​LinkedBlockingDeque​​是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移出元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。
  多了​​addFirst​​、​​addLast​​、​​offerFirst​​、​​offerLast​​、​​peekFirst​​和​​peekLast​​等方法，以​​First​​单词结尾的方法，表示插入、获取（​​peek​​）或移除双端队列的第一个元素。以​​Last​​单词结尾的方法，表示插入、获取或移除双端队列的最后一个元素。另外，插入方法​​add​​等同于​​addLast​​，移除方法​​remove​​等效于​​removeFirst​​。但是​​take​​方法却等同于​​takeFirst​​，不知道是不是JDK的bug，使用时还是用带有​​First​​和​​Last​​后缀的方法更清楚。在初始化​​LinkedBlockingDeque​​时可以设置容量防止其过度膨胀。另外，双向阻塞队列可以运用在“工作窃取”模式中。
• 注意事项：
  1.​​Array​​实现和​​Linked​​实现的区别
  ① 队列中锁的实现不同
  ​​ArrayBlockingQueue​​实现的队列中的锁是没有分离的，即生产和消费用的是同一个锁；
  ​​LinkedBlockingQueue​​实现的队列中的锁是分离的，即生产用的是​​putLock​​，消费是​​takeLock​​ ② 在生产或消费时操作不同
  ​​ArrayBlockingQueue​​实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的；
  ​​LinkedBlockingQueue​​实现的队列中在生产和消费的时候，需要把枚举对象转换为​​Node<E>​​进行插入或移除，会影响性能
  ③队列大小初始化方式不同
  ​​ArrayBlockingQueue​​实现的队列中必须指定队列的大小；
  ​​LinkedBlockingQueue​​实现的队列中可以不指定队列的大小，但是默认是​​Integer.MAX_VALUE​​ 2.有界无界？
  有界队列就是长度有限，满了以后生产者会阻塞，无界队列就是里面能放无数的东西而不会因为队列长度限制被阻塞，当然空间限制来源于系统资源的限制，如果处理不及时，导致队列越来越大越来越大，超出一定的限制致使内存超限，操作系统或者JVM帮你解决烦恼，直接把你 OOM kill 省事了。
  无界也会阻塞，为何？因为阻塞不仅仅体现在生产者放入元素时会阻塞，消费者拿取元素时，如果没有元素，同样也会阻塞。

二.线程池

1.为啥要用线程池

①降低资源消耗。通过​​重复利用已创建的线程​​​降低线程创建和销毁造成的消耗。
②提高响应速度。当任务到达时，任务可以​​不需要等到线程创建​​​就能立即执行。
假设一个服务器完成一项任务所需时间为：T1 创建线程时间，T2 在线程中执行任务的时间，T3 销毁线程时间。 如果：T1 + T3 远大于 T2，则可以采用线程池，以提高服务器性能。线程池技术正是关注如何缩短或调整T1,T3时间的技术，从而提高服务器程序性能的。它把T1，T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时，不会有T1，T3的开销了。而我们自己​​​new​​​线程的时候就是一个任务需要创建一个线程然后执行任务还要销毁线程。这个时间效率是远远不如线程池的。
③提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。

2.线程池的生活例子

我先来口头说一下线程池的工作机制。就是线程池里面全是线程，来了任务之后就会从线程池里面分配一个线程来处理这个任务。当任务够多时，超过了线程池中线程的正常数目，则会去阻塞队列中等待。当然，阻塞队列也是有容量的。当任务再次足够多的时候就会到达阻塞队列的上限，然后就会创建新线程来处理任务。当然创建新线程的数目是有限的，当任务又超过这个上限时，此时就会有一种“拒绝”机制来拒绝新来的这个任务。整体来说就是这样。

我们来举一个生活中的例子，就是银行。

最开始，有6个任务，我标出了序号。前三个任务由于来的比较早，所以去柜台处理任务了。柜台就是线程池，里面在正常情况下有三个线程来处理任务。

当任务足够多（超过6个时）的时候，就会创建四号线程。当再来一个任务的时候，就会创建五号线程。如图

此时已经到达线程池的上限了。如果再来任务，则会实行拒绝机制

这就是线程池的基本原理。下面我用代码来实现一下这个原理。

3.线程池举例

public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(3, 5, 1L, TimeUnit.SECONDS
                , new ArrayBlockingQueue<>(3), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());

        for (int i = 0; i < 9; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() +  "在办理业务");
                }
            });
        }
    }

一开始有3个任务的时候，就是阻塞队列为空，123线程分别执行一次。

后来任务大于3小于等于6的时候，会有一个线程执行两次的情况。这里按6来举例。

当线程数为7时，会创建线程4

当线程数为8时，会创建线程5

当线程数为9当时候会执行拒绝策略，抛出异常

4.线程池实体类介绍

我们在上面的例子中也知道了线程池当实体类为​​ThreadPoolExecutor​​​，那么与它关联的一些类都与它有什么关系呢？
​​​Executor​​​是一个接口，它是​​Executor​​​框架的基础，它将任务的提交与任务的执行分离开来。
​​​ExecutorService​​​接口继承了​​Executor​​​，在其上做了一些​​shutdown()、submit()​​​的扩展，可以说是真正的线程池接口；
​​​AbstractExecutorService​​​抽象类实现了​​ExecutorService​​​接口中的大部分方法；
​​​ThreadPoolExecutor​​​是线程池的核心实现类，用来执行被提交的任务。继承了​​AbstractExecutorService​​

线程池创建的各个参数含义

①​​corePoolSize​​就是核心线程数，就是正常的线程数目

②​​maximumPoolSize​​就是最大线程数，就是上面银行例子的 5

③​​keepAliveTime​​​是 后来创建的线程没有事情做时的最大存活时间。一般来说这个参数只有线程数大于​​corePoolSize​​的时候才有用

④​​TimeUnit​​​是​​keepAliveTime​​的时间单位

⑤​​threadFactory​​是线程工厂，即创建线程的工厂，一般来说用默认的就可以

⑥​​RejectedExecutionHandler​​ 线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：

（1）​​AbortPolicy​​：直接抛出异常，默认策略；

（2）​​CallerRunsPolicy​​：用调用者所在的线程来执行任务；

（3）​​DiscardOldestPolicy​​：丢弃阻塞队列中靠最前的任务，并执行当前任务；

（4）​​DiscardPolicy​​：直接丢弃任务；

当然也可以根据应用场景实现​​RejectedExecutionHandler​​接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。

5.线程池工作机制

1）如果当前运行的线程少于​​corePoolSize​​​，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。
2）如果运行的线程等于或多于​​​corePoolSize​​​，则将任务加入​​BlockingQueue​​​。
3）如果无法将任务加入​​​BlockingQueue​​​（队列已满），则创建新的线程来处理任务。
4）如果创建新线程将使当前运行的线程超出​​​maximumPoolSize​​​，任务将被拒绝，并调用​​RejectedExecutionHandler.rejectedExecution()​​方法。

6.提交任务的​​execute​​​和​​submit​​的两个区别

①​​execute()​​​方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功。
②​​​submit()​​​方法用于提交需要返回值的任务。线程池会返回一个​​future​​​类型的对象，通过这个​​future​​​对象可以判断任务是否执行成功，并且可以通过​​future​​​的​​get​​​方法来获取返回值，​​get​​​方法会阻塞当前线程直到任务完成，而使用​​get（long timeout，TimeUnit unit）​​方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。

7.如何关闭线程池

可以通过调用线程池的​​shutdown​​​或​​shutdownNow​​​方法来关闭线程池。它们的原理是遍历线程池中的工作线程，然后逐个调用线程的​​interrupt​​方法来中断线程，所以无法响应中断的任务可能永远无法终止。但是它们存在一定的区别，
①​​​shutdownNow​​​首先将线程池的状态设置成​​STOP​​，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表，而②​​shutdown​​只是将线程池的状态设置成​​SHUTDOWN​​状态，然后中断所有没有正在执行任务的线程。
只要调用了这两个关闭方法中的任意一个，​​​isShutdown​​​方法就会返回​​true​​​。当所有的任务都已关闭后，才表示线程池关闭成功，这时调用​​isTerminaed​​​方法会返回​​true​​​。
至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用​​​shutdown​​​方法来关闭线程池，如果任务不一定要执行完，则可以调用​​shutdownNow​​方法。

8.如何合理配置线程池

要想合理地配置线程池，就必须首先分析任务特性，可以从以下几个角度来分析。

• 任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
  任务的优先级：高、中和低。
  任务的执行时间：长、中和短。
  任务的依赖性：是否依赖其他系统资源，如数据库连接。
  性质不同的任务可以用不同规模的线程池分开处理。

①CPU密集型任务应配置尽可能小的线程，如配置CPU核心数+1个线程的线程池。由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2*CPU核心数。
混合型的任务，如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行分解。可以通过​​Runtime.getRuntime().availableProcessors()​​​方法获得当前设备的CPU个数。
②优先级不同的任务可以使用优先级队列​​​PriorityBlockingQueue​​​来处理。它可以让优先级高的任务先执行。
③执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行。
建议使用有界队列。有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点儿，比如几千。如果当时我们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存，导致整个系统不可用，而不只是后台任务出现问题。