Java并发编程之阻塞队列与Fork/Join框架

谁借莪１个温暖的怀抱￠ 2022-04-10 11:00 131阅读 0赞

## 阻塞队列 ##

阻塞队列是支持两个附加操作的队列。这两个附加操作支持阻塞插入和移除方法：

*  阻塞启用的插入：当队列满时，它会阻塞插入元素的线程，直到队列不满意为止。
 *  阻塞删除：当队列为空时，检索元素的线程等待队列变为非空。

阻塞队列通常用于生产者-消费者场景。生产者是向队列添加元素的线程，使用者是将元素从队列中取出的线程，阻塞队列是存储和检索元素的容器。

阻塞队列的4种处理方式：

1.  抛出异常：
    
     *  `add(e)` 当队列满，再插入元素，抛出异常
     *  `remove()` 当队列空，再删除元素，抛出异常
     *  `element()` 获取元素
2.  返回特殊值：
    
     *  `offer(e)` 插入元素时，插入成功返回true
     *  `poll()` 移除元素，成功返回该值，否则返回null
     *  `peek()`
3.  一直阻塞：
    
     *  `put(e)` 当阻塞队列满时，再插入时会阻塞生产者线程，直到队列可用或中断退出
     *  `take()` 当队列空，再移除元素会阻塞消费者线程，直到队列不空
4.  超时退出
    
     *  `offer(e, time, unit)` 队列满时再插入元素，阻塞，超时退出
     *  `poll(time, unit)` 队列空时移除元素，阻塞，超时退出

### Java中几种阻塞队列 ###

*  ArrayBlockingQueue: 数组结构构成的有界 FIFO 阻塞队列
 *  LinkedBolckingQueue: 链表结构构成的有界 FIFO 阻塞队列
 *  PriorityBlockingQueue: 支持优先级排序的无界阻塞队列
 *  DelayQueue: 支持延时获取元素，使用优先级队列实现的无界阻塞队列
 *  SynchronousQueue: 不存储元素的阻塞队列，不为队列元素维护存储空间
 *  LinkedTransferQueue: 链表结构构成的无界阻塞队列
 *  LinkedBlockingDeque: 链表构成的双向阻塞队列

ArrayBlockingQueue

数组阻塞队列是一个使用数组实现的有界队列，数组根据FIFO原则对元素进行排序。它还支持将生产者线程和使用者线程排队的可选公平性策略，这不保证默认情况下对线程的公平访问，并且可以公平地构造。公平减少了吞吐量，但减少了可变性，并避免了“不平衡”。

LinkedBlockingQueue

这是一个使用链表实现的有界阻塞队列。默认长度和最大长度为整数。最大值。队列还根据FIFO原则对元素进行排序，以确定线程执行的顺序。

PriorityBlockingQueue

这是一个支持优先级的无边界的zusu队列。默认情况下，采用自然升序，也可以通过构造函数指定comparator对元素进行排序。但它不能保证相同优先元素的顺序。

底层是采用二叉最大堆来实现优先级排序的。

DelayQueue

这是一个支持延迟获取元素的无边界阻塞队列，其队列使用优先级队列PriorityQueue实现。队列中的元素必须实现延迟的接口。创建元素时，可以指定从队列中检索元素所需的时间，并且只能在元素过期时指定。

主要用于缓存，例如清除缓冲区中超时的数据。它还用于调度定时任务。

创建元素时，首先初始化延迟的接口；然后实现getDelay（TimeUnit Unit）方法，返回的值是当前元素需要延迟多长时间；最后实现compareTo（delayed other）方法以指定元素的顺序。

当使用者从队列中检索元素时，如果元素未达到延迟时间，则当前线程将被阻塞。此外，leader变量被设置为表示等待获取队列头元素的线程。如果前导不为空，则表示已经准备好等待获取队列头元素，并且使用wait（）方法使当前线程等待信号。如果leader为空，则将当前线程设置为leader，并使用waitnanos（）方法使当前线程等待接收到的信号或延迟时间。

SynchronousQueue

与其他阻塞队列不同，这是一个不存储元素的阻塞队列。每个Put操作必须等待Take操作，否则不能继续添加元素，反之亦然。它分为公平访问队列和不公平访问队列，默认情况下由不公平策略访问。

队列本身不存储任何元素，并且适用于可传递的场景。它将生产者线程处理的数据直接传输到使用者线程。它的吞吐量高于链接阻塞队列和数组阻塞队列。

LinkedTransferQueue

这是由链表结构组成的FIFO的无边界阻塞传输队列。它采用先发制人的方式，即有先发制人的方式时，直接采取先发制人的方式，并占据该位置，直到得到该位置，超时或中断为止。与其他阻塞队列相比，有更多的TryTransfer方法和传输方法。

*  `transfer(e,[timeout,unit])` 方法: 如果当前有消费者正等待接收元素，该方法可以把生产者传入的元素立刻传输给消费者。如果没有消费者等待，该方法将元素存放在队列的 tail 节点，等到该元素被消费者消费了才返回。
 *  `tryTransfer(e,[timeout,unit])`方法： 试探生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，返回false。该方法无论消费者是否接收都立即返回，而 transfer 方法必须等消费了才返回。

LinkedBlockingDeque

是一个由链表组成的双向阻塞队列。可以从队列两端插入和移除元素。

## Fork/Join框架 ##

该框架主要用于并行计算，将一个大的人分成几个小的任务，最后对每个小任务的结果进行汇总，得到框架的结果。fork将一个大任务分成几个子任务，并并行执行它们。join是合并这些子任务并最终得到大任务结果的结果。

### 工作窃取算法 ###

窃取工作是指从其他队列中窃取要执行的任务的线程。通常使用两端队列。被盗任务线程总是从两端队列的头部获取要执行的任务，被盗任务线程总是从两端队列的尾部获取要执行的任务。

其优点是充分利用线程进行并行计算，减少了线程之间的竞争。缺点是在某些情况下存在竞争，例如当队列只有一个任务时，它将消耗更多的资源。

### 框架设计思路 ###

首先，将任务划分为子任务，并将子任务连续划分，直到子任务足够小为止。

然后，执行任务并合并结果。分离的子任务被放置在双端队列中，然后几个启动线程从双端队列获得任务执行。执行结果放在队列中，启动一个线程从队列中获取数据，并合并这些线程。

### 示例 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkMDkwNDQ5MDFndWlj_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NkMDkwNDQ5MDFndWlj_size_16_color_FFFFFF_t_70]: /images/20220331/da0a71ce6dcc4bdd8c58da7d9573886b.png