阻塞队列介绍

╰+哭是因爲堅強的太久メ 2024-03-16 10:17 58阅读 0赞

下面我们先介绍阻塞队列。

**阻塞****队****列****:**  
阻塞队列 (BlockingQueue)是Java util.concurrent包下重要的数据结构，BlockingQueue提供了线程安全的队列访问方式：当阻塞队列进行插入数据时，如果队列已满，线程将会阻塞等待直到队列非满；从阻塞队列取数据时，如果队列已空，线程将会阻塞等待直到队列非空。并发包下很多高级同步类的实现都是基于BlockingQueue实现的。

➢ **Block****i****ngQueu****e****阻塞队****列**

BlockingQueue 通常用于一个线程生产对象，而另外一个线程消费这些对象的场景。下图是对这个原理的阐述：

Thread 1 Thread 2

**B****l****o****c****ki****n****g****Q****u****e****u****e**

Put Take

一个线程往里边放，另外一个线程从里边取的一个 BlockingQueue。

一个线程将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限的。如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中，直到负责消费的线程从队列中拿走一个对象。负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试去从一个空的队列中提取对象的话，这个消费线程将会处于阻塞之中，直到一个生产线程把一个对象丢进队列。

**Blo****c****kin****g****Q****u****e****u****e****的****方法****:**  
BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：

**阻塞****队****列****提****供****了****四****种****处****理****方****法****:**

<table style="margin-left:23.1pt;"> 
 <tbody> 
 <tr> 
 <td style="background-color:#f1f1f1;width:122.2pt;"> 方法\处理方式&nbsp; </td> 
 <td style="background-color:#f1f1f1;width:92.3pt;"> 抛出异常&nbsp; </td> 
 <td style="background-color:#f1f1f1;width:103.9pt;"> 返回特殊值&nbsp; </td> 
 <td style="background-color:#f1f1f1;width:89.3pt;"> 一直阻塞&nbsp; </td> 
 <td style="background-color:#f1f1f1;width:136.7pt;"> 超时退出&nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:122.2pt;"> 插入方法&nbsp; </td> 
 <td style="vertical-align:top;width:92.3pt;"> add(e)&nbsp; </td> 
 <td style="vertical-align:top;width:103.9pt;"> offer(e)&nbsp; </td> 
 <td style="vertical-align:top;width:89.3pt;"> put(e) </td> 
 <td style="vertical-align:top;width:136.7pt;"> offer(e,time,unit)&nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:122.2pt;"> 移除方法&nbsp; </td> 
 <td style="vertical-align:top;width:92.3pt;"> remove()&nbsp; </td> 
 <td style="vertical-align:top;width:103.9pt;"> poll()&nbsp; </td> 
 <td style="vertical-align:top;width:89.3pt;"> take() </td> 
 <td style="vertical-align:top;width:136.7pt;"> poll(time,unit)&nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;width:122.2pt;"> 检查方法&nbsp; </td> 
 <td style="vertical-align:top;width:92.3pt;"> element() </td> 
 <td style="vertical-align:top;width:103.9pt;"> peek()&nbsp; </td> 
 <td style="vertical-align:top;width:89.3pt;"> 不可用&nbsp; </td> 
 <td style="vertical-align:top;width:136.7pt;"> 不可用&nbsp; </td> 
 </tr> 
 </tbody> 
</table>

**四组****不****同****的****行****为****方****式****解****释****：**

抛异常：如果试图的操作无法立即执行，抛一个异常。

特定值：如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。

阻塞：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。

超时：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定 值。返回一个特定值以告知该操作是否成功(典型的是true / false)。

**无法向****一****个****B****l****o****c****kin****g****Q****ueue****中****插****入****n****u****l****l****。****如果你****试图****插****入****n****ul****l****，****B****l****oc****k****ing****Q****ueu****e****将****会****抛出****一****个****N****ul****l****P****o****i****n****t****erE****x****c****eption****.**

**Blo****c****kin****g****Q****u****e****u****e****的****实****现****类****:**  
BlockingQueue 是个接口，你需要使用它的实现之一来使用BlockingQueue ，[J][][a][J][v][J][a][J][.][J]util.concurrent包下具有以下 BlockingQueue 接口的实现类：

l ArrayBlockingQueue：ArrayBlockingQueue 是一个有界的阻塞队列，其内部实现是将对象放到一个数组里。有界也就意味着，它不能够存储无限多数量的元素。它有一个同一时间能够存储元素数量的上限。你可以在对其初始化的时候设定这个上限，但之后就无法对这个上限进行修改了(译者注：因为它是基于数组实现的，也就具有数组的特性：一旦初始化，大小就无法修改)。

l DelayQueue：DelayQueue 对元素进行持有直到一个特定的延迟到期。注入其中的元素必须实现 java.util.concurrent.Delayed 接口。

l LinkedBlockingQueue：LinkedBlockingQueue 内部以一个链式结构(链接节点)对其元素进行存储。如果需要的话，这一链式结构可以选择一个上限。如果没有定义上限，将使用 Integer.MAX\_VALUE 作为上限。

l PriorityBlockingQueue： PriorityBlockingQueue 是一个 无界的并发队列 。它使用了和类 java.util.PriorityQueue 一样的排序规则。你无法向这个队列中插入 null 值。所有插入到 PriorityBlockingQueue 的元素必须实现 java.lang.Comparable 接口。因此该队列中元素的排序就取决于你自己的 Comparable 实现。

l SynchronousQueue：SynchronousQueue 是一个特殊的队列，它的内部同时只能够容纳单个元素。如果该队列已有一元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。据此，把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。

➢ **A****rr****ayB****l****oc****k****ingQue****u****e****阻塞队列**

**A****rr****a****y****B****lo****c****king****Q****ueu****e****类****图**

如上图 ArrayBlockingQueue 内部有个数组 items 用来存放队列元素，putindex 下标标示入队元素下标，

takeIndex是出队下标，count统计队列元素个数，从定义可知道并没有使用volatile修饰，这是因为访问这些变量使

用都是在锁块内，并不存在可见性问题。另外有个独占锁lock用来对出入队操作加锁，这导致同时只有一个线程可以

访问入队出队，另外notEmpty，notFull条件变量用来进行出入队的同步。

另外构造函数必须传入队列大小参数，所以为有界队列，默认是Lock为非公平锁。

public ArrayBlockingQueue(int capacity) \{

this(capacity, false);

public ArrayBlockingQueue(int capacity, boolean fair) \{

if (capacity <= 0)

throw new IllegalArgumentException();

this.items = new Object\[capacity\];

lock = new ReentrantLock(fair);

notEmpty = lock.newCondition();

notFull = lock.newCondition();

**ps****：**

**所谓****公****平****锁****：**就是在并发环境中，每个线程在获取锁时会先查看此锁维护的等待队列，如果为空，或者当前线

程线程是等待队列的第一个，就占有锁，否则就会加入到等待队列中，以后会按照FIFO的规则从队列中取到

自己。

**非公****平****锁****：**比较粗鲁，上来就直接尝试占有锁，如果尝试失败，就再采用类似公平锁那种方式

**A****rr****a****y****B****lo****c****king****Q****ueu****e****方****法**

✓ **of****fe****r****方****法**

在队尾插入元素，如果队列满则返回false，否者入队返回true。

public boolean offer(E e) \{

//e为null，则抛出NullPointerException异常

checkNotNull(e);

//获取独占锁

final ReentrantLock lock = this.lock;

lock.lock();

try \{

//如果队列满则返回false

if (count == items.length)

return false;

else \{

//否者插入元素

insert(e);

return true;

\} finally \{

//释放锁

lock.unlock();

private void insert(E x) \{

//元素入队

items\[putIndex\] = x;

//计算下一个元素应该存放的下标

putIndex = inc(putIndex);

\++count;

notEmpty.signal();

//循环队列，计算下标

final int inc(int i) \{

return (++i == items.length) ? 0 : i;

这里由于在操作共享变量前加了锁，所以不存在内存不可见问题，加过锁后获取的共享变量都是从主内存获取的，

而不是在CPU缓存或者寄存器里面的值，释放锁后修改的共享变量值会刷新会主内存中。

另外这个队列是使用循环数组实现，所以计算下一个元素存放下标时候有些特殊。另外 insert 后调用

notEmpty.signal();是为了激活调用notEmpty.await()阻塞后放入notEmpty条件队列中的线程。

✓ **Pu****t****操****作**

在队列尾部添加元素，如果队列满则等待队列有空位置插入后返回。

public void put(E e) throws InterruptedException \{

checkNotNull(e);

final ReentrantLock lock = this.lock;

//获取可被中断锁

lock.lockInterruptibly();

try \{

//如果队列满，则把当前线程放入notFull管理的条件队列

while (count == items.length)

notFull.await();

//插入元素

insert(e);

\} finally \{

lock.unlock();

需要注意的是如果队列满了那么当前线程会阻塞，知道出队操作调用了 notFull.signal 方法激活该线程。代码逻

辑很简单，但是这里需要思考一个问题为啥调用 lockInterruptibly 方法而不是 Lock 方法。我的理解是因为调用了条

件变量的await()方法，而await()方法会在中断标志设置后抛出InterruptedException异常后退出，所以还不如在加

锁时候先看中断标志是不是被设置了，如果设置了直接抛出InterruptedException异常，就不用再去获取锁了。然后

看了其他并发类里面凡是调用了await 的方法获取锁时候都是使用的 lockInterruptibly 方法而不是Lock 也验证了这

个想法。

✓ **P****ol****l****操****作**

从队头获取并移除元素，队列为空，则返回null。

public E poll() \{

final ReentrantLock lock = this.lock;

lock.lock();

try \{

//当前队列为空则返回null,否者

return (count == 0) ? null : extract();

\} finally \{

lock.unlock();

private E extract() \{

final Object\[\] items = this.items;

//获取元素值

E x = this.<E>cast(items\[takeIndex\]);

//数组中值值为null;

items\[takeIndex\] = null;

//队头指针计算，队列元素个数减一

takeIndex = inc(takeIndex);

\--count;

//发送信号激活notFull条件队列里面的线程

notFull.signal();

return x;

✓ **T****a****k****e****操****作**

从队头获取元素，如果队列为空则阻塞直到队列有元素。

public E take() throws InterruptedException \{

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try \{

//队列为空，则等待，直到队列有元素

while (count == 0)

notEmpty.await();

return extract();

\} finally \{

lock.unlock();

需要注意的是如果队列为空，当前线程会被挂起放到 notEmpty 的条件队列里面，直到入队操作执行调用

notEmpty.signal后当前线程才会被激活，await才会返回。

✓ **P****ee****k****操****作**

返回队列头元素但不移除该元素，队列为空，返回null。

public E peek() \{

final ReentrantLock lock = this.lock;

lock.lock();

try \{

//队列为空返回null,否者返回头元素

return (count == 0) ? null : itemAt(takeIndex);

\} finally \{

lock.unlock();

final E itemAt(int i) \{

return this.<E>cast(items\[i\]);

✓ **Siz****e****操作**

获取队列元素个数，非常精确因为计算size时候加了独占锁，其他线程不能入队或者出队或者删除元素。

public int size() \{

final ReentrantLock lock = this.lock;

lock.lock();

try \{

return count;

\} finally \{

lock.unlock();

✓ **A****rr****a****y****B****lo****c****king****Q****ueu****e****小****结**

ArrayBlockingQueue 通过使用全局独占锁实现同时只能有一个线程进行入队或者出队操作，这个锁的粒度比较

大，有点类似在方法上添加synchronized的意味。其中offer,poll操作通过简单的加锁进行入队出队操作，而put,take

则使用了条件变量实现如果队列满则等待，如果队列空则等待，然后分别在出队和入队操作中发送信号激活等待线程

实现同步。另外相比 LinkedBlockingQueue，ArrayBlockingQueue 的 size 操作的结果是精确的，因为计算前加了

全局锁。

**A****rr****a****y****B****lo****c****kin****g****Q****ueu****e****示****例**

需求：在多线程操作下，一个数组中最多只能存入 3 个元素。多放入不可以存入数组，或等待某线程对数组中某

个元素取走才能放入，要求使用java的多线程来实现。（面试）

代码实现：

public class BlockingQueueTest \{

public static void main(String\[\] args) \{

final BlockingQueue queue = new ArrayBlockingQueue(3);

for(int i=0;i<2;i++)\{

new Thread()\{

public void run()\{

while(true)\{

try \{

Thread.sleep((long)(Math.random()\*1000));  
System.out.println(Thread.currentThread().getName() + "准备放数据!");

queue.put(1);  
System.out.println(Thread.currentThread().getName() + "已经放了数据，" +

"队列目前有" + queue.size() + "个数据");

\} catch (InterruptedException e) \{

e.printStackTrace();

\}.start();

new Thread()\{

public void run()\{

while(true)\{

try \{

//将此处的睡眠时间分别改为100和1000，观察运行结果

Thread.sleep(100);  
System.out.println(Thread.currentThread().getName() + "准备取数据!"); System.err.println(queue.take());  
System.out.println(Thread.currentThread().getName() + "已经取走数据，" +

"队列目前有" + queue.size() + "个数据");

\} catch (InterruptedException e) \{

e.printStackTrace();

\}.start();

输出结果：

Thread-0准备放数据!

Thread-0已经放了数据，队列目前有1个数据

Thread-0准备放数据!

Thread-0已经放了数据，队列目前有2个数据

Thread-1准备放数据!

Thread-1已经放了数据，队列目前有3个数据

Thread-2准备取数据!