4种线程池和7种并发队列

Posted 2023-03-18

参考技术A Java并发包中的阻塞队列一共7个，当然他们都是线程安全的。 

ArrayBlockingQueue：一个由数组结构组成的 有界 阻塞队列。 

LinkedBlockingQueue：一个由链表结构组成的无界阻塞队列。 

PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。 

DealyQueue：一个使用优先级（启动时间）队列实现的无界阻塞队列。 

SynchronousQueue：一个不存储元素的阻塞队列。 

LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。 

LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列。

常用的只有三个，重点是前两个

LinkedBlockingQueue和ArrayBlockingQueue区别：

1、LinkedBlockingQueue内部由两个ReentrantLock来实现出入队列的线程安全，由各自的Condition对象的await和signal来实现等待和唤醒功能。而ArrayBlockingQueue的只使用一个ReentrantLock管理进出队列。

而LinkedBlockingQueue实现的队列中的锁是分离的，其添加采用的是putLock，移除采用的则是takeLock，这样能大大提高队列的吞吐量，也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

2、队列大小有所不同，ArrayBlockingQueue是有界的初始化必须指定大小，而LinkedBlockingQueue可以是有界的也可以是无界的(Integer.MAX_VALUE)，对于后者而言，当添加速度大于移除速度时，在无界的情况下，可能会造成内存溢出等问题。

3、数据存储容器不同，ArrayBlockingQueue采用的是数组作为数据存储容器，而LinkedBlockingQueue采用的则是以Node节点作为连接对象的链表。

由于ArrayBlockingQueue采用的是数组的存储容器，因此在插入或删除元素时不会产生或销毁任何额外的对象实例，而LinkedBlockingQueue则会生成一个额外的Node对象。这可能在长时间内需要高效并发地处理大批量数据的时，对于GC可能存在较大影响。

SynchronousQueue

没有容量，是无缓冲等待队列，是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素

使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作。

1、newfixed，线程默认大小确定、最大大小确定（实际没什么用），默认使用linkedblockqueue,无尽队列

危害在于这个等待队列，队列如果消费不及时不断膨胀可以使机器资源耗尽

ArrayBlockingQueue是一个有界缓存等待队列，可以指定缓存队列的大小，当正在执行的线程数等于corePoolSize时，多余的元素缓存在ArrayBlockingQueue队列中等待有空闲的线程时继续执行，当ArrayBlockingQueue已满时，加入ArrayBlockingQueue失败，会开启新的线程去执行，当线程数已经达到最大的maximumPoolSizes时，再有新的元素尝试加入ArrayBlockingQueue时会报错。

2、cached，线程数不限大小

危害 本身就是没有限制，有多少请求创建多少线程，直到资源耗尽

CachedThreadPool使用没有容量的SynchronousQueue作为主线程池的工作队列，它是一个没有容量的阻塞队列。每个插入操作必须等待另一个线程的对应移除操作。这意味着，如果主线程提交任务的速度高于线程池中处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下，CachedThreadPool会因为创建过多线程而耗尽CPU资源。

3、single

可顺序执行任务，同时只有一个线程处理（单线程）

执行过程如下：

1.如果当前工作中的线程数量少于corePool的数量，就创建一个新的线程来执行任务。

2.当线程池的工作中的线程数量达到了corePool，则将任务加入LinkedBlockingQueue。

3.线程执行完1中的任务后会从队列中去任务。

注意：由于在线程池中只有一个工作线程，所以任务可以按照添加顺序执行。

4、ScheduledThreadPool

public ScheduledThreadPoolExecutor(int corePoolSize)

        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,

              new DelayedWorkQueue());

   

使用了延迟队列，无界，和cached类似

如果运行的线程达到了corePoolSize，就把任务添加到任务队列DelayedWorkQueue中；DelayedWorkQueue会将任务排序，先执行的任务放在队列的前面。

任务执行完后，ScheduledFutureTask中的变量time改为下次要执行的时间，并放回到DelayedWorkQueue中

DelayQueue是一个没有边界BlockingQueue实现，加入其中的元素必需实现Delayed接口。 当生产者线程调用put之类的方法加入元素时，会触发Delayed接口中的compareTo方法进行排序，也就是说队列中元素的顺序是按到期时间排序的，而非它们进入队列的顺序。排在队列头部的元素是最早到期的，越往后到期时间赿晚。

消费者线程查看队列头部的元素，注意是查看不是取出。然后调用元素的getDelay方法，如果此方法返回的值小0或者等于0，则消费者线程会从队列中取出此元素，并进行处理。如果getDelay方法返回的值大于0，则消费者线程wait返回的时间值后，再从队列头部取出元素，此时元素应该已经到期。

DelayQueue是Leader-Followr模式的变种，消费者线程处于等待状态时，总是等待最先到期的元素，而不是长时间的等待。消费者线程尽量把时间花在处理任务上，最小化空等的时间，以提高线程的利用效率。

无论创建那种线程池 必须要调用ThreadPoolExecutor，以上四种都是调用这个实现的

线程池类为 java.util.concurrent.ThreadPoolExecutor，常用构造方法为：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,

long keepAliveTime, TimeUnit unit,

BlockingQueue workQueue,

RejectedExecutionHandler handler)

corePoolSize： 线程池维护线程的最少数量 

maximumPoolSize：线程池维护线程的最大数量 

keepAliveTime： 线程池维护线程所允许的空闲时间 ， unit： 线程池维护线程所允许的空闲时间的单位 

workQueue： 线程池所使用的缓冲队列 

handler： 线程池对拒绝任务的处理策略 --饱和策略

-------------------------------------------------------------------

通用流程：

一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是 Runnable类型对象的run()方法。

当一个任务通过execute(Runnable)方法欲添加到线程池时：

如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

如果此时线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。

如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。

也就是：处理任务的优先级为：

核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。 

当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

---------------------------------------------------------------------------------------------

unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：

NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。

workQueue我常用的是：java.util.concurrent. ArrayBlockingQueue

handler有四个选择：

1、【抛异常】ThreadPoolExecutor.AbortPolicy() 抛出java.util.concurrent.RejectedExecutionException异常

2、【重试】ThreadPoolExecutor.CallerRunsPolicy() 重试添加当前的任务，他会自动重复调用execute()方法

ThreadPoolExecutor.DiscardOldestPolicy()

3、【找个旧的停了】抛弃旧的任务 ThreadPoolExecutor.DiscardPolicy()

4、【不抛异常】抛弃当前的任务

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

https://www.cnblogs.com/baizhanshi/p/5469948.html