Java并发阻塞队列BlockingQueue概览

不念不忘少年蓝@ 2022-06-12 11:39 243阅读 0赞

## **BlockingQueue** ##

`java.util.concurrent`包下的 **BlockingQueue** 接口规范了一个放数据、取数据都是线程安全的队列。

### **BlockingQueue的用法** ###

一个典型的使用 BlockingQueue 的应用场景是 生产-消费者模型。  
![在这里插入图片描述][20180919155809666]

**一个线程往队列生产数据，另一个队列消耗数据**

生产线程会持续生产新的实例放入队列中，直到队列容量达到指定值。也就是说，如果队列数据大小达到队列可以容纳的上限，那么生产线程会一致阻塞。直到消费线程开始消耗队列中的数据。

消费线程会持续从队列获取数据，直到队列为空就会阻塞，直至有生产线程方数据到队列中。

### **BlockingQueue 的常用方法** ###

BlockingQueue 有4组不同的方法插入、移除、检测是否存在元素的方法。每一组方法的表现行为都不一样。

<table> 
 <thead> 
  <tr> 
   <th>&nbsp;</th> 
   <th align="left">抛异常</th> 
   <th>返回指定值(一般是布尔值)</th> 
   <th>阻塞</th> 
   <th>超时</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>插入</strong></td> 
   <td align="left"><font>add(o)</font></td> 
   <td><font>offer(o)</font></td> 
   <td><font>put(o)</font></td> 
   <td>offer(o, timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td><strong>移除</strong></td> 
   <td align="left"><font>remove(o)</font></td> 
   <td><font>poll()</font></td> 
   <td><font>take()</font></td> 
   <td>poll(timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td><strong>检测元素</strong></td> 
   <td align="left"><font>element()</font></td> 
   <td><font>peek()</font></td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

4组不同表现行为的说明：

*  **抛异常Throws Exception**: 如果指定的操作不能马上执行，则会抛出异常
 *  **返回指定值**： 如果指定的操作不能马上执行，则会返回指定的值，一般是true / false
 *  **阻塞**： 如果指定的操作不能马上执行，则会一直阻塞
 *  **超时**： 如果指定的操作不能马上执行，则会阻塞直到给定的时间单位，最后返回一个表示操作是否成功执行的标志（一般是true / false）

不允许往BlockingQueue里赛空值NULL，否则抛出NullPointerException。

并不是只能访问BlockingQueue的对头、对尾元素，还可以对其他元素进行操作。例如你可以使用 remove(o)去取消使用元素o，但是并不建议这样使用因为效率并不高。

### **BlockingQueue的实现类** ###

因为 BlockingQueue 是一个接口，所以你需要使用它的实现类来使用。`java.util.concurrent`包有以下几个常用的实现类：

*  **ArrayBlockingQueue**
 *  **DelayQueue**
 *  **LinkedBlockingQueue**
 *  **PriorityBlockingQueue**
 *  **SynchronousQueue**

这几个类的使用我们在后面节中讨论，你也可以先通过查看JAVA DOC先了解使用。

### **Java BlockingQueue 的使用实例** ###

我们使用 ArrayBlockingQueue 这个实现类来使用java中的阻塞队列。

首先， BlockingQueueExample 类开启了生产者、消费者两个不同的线程。 Producer 插入字符串到共享的队列中去。Consumer 就从队列中取数据。

public class BlockingQueueExample { public static void main(String[] args) throws Exception { BlockingQueue queue = new ArrayBlockingQueue(1024); Producer producer = new Producer(queue); Consumer consumer = new Consumer(queue); new Thread(producer).start(); new Thread(consumer).start(); Thread.sleep(4000); } }

这是 Producer 类。 注意：生产者在 put方法后休眠了1秒钟。这是为了演示让Consumer阻塞，直到等到Producer放入新值到队列中去。

public class Producer implements Runnable{ protected BlockingQueue queue = null; public Producer(BlockingQueue queue) { this.queue = queue; } public void run() { try { queue.put("1"); Thread.sleep(1000); queue.put("2"); Thread.sleep(1000); queue.put("3"); } catch (InterruptedException e) { e.printStackTrace(); } } }

Consumer 类仅仅从队列中取数据，并打印输出。

public class Consumer implements Runnable{ protected BlockingQueue queue = null; public Consumer(BlockingQueue queue) { this.queue = queue; } public void run() { try { System.out.println(queue.take()); System.out.println(queue.take()); System.out.println(queue.take()); } catch (InterruptedException e) { e.printStackTrace(); } } }

## **有界阻塞队列ArrayBlockingQueue** ##

`ArrayBlockingQueue`是 BlockingQueue 接口的一个实现。它是一个 有界的、阻塞队列，数据元素通过一个内部数据存储。

*  有界 ： 指的是不能存储无限的元素。它有一个上限，这应该在初始化时指定。

`ArrayBlockingQueue`中的元素读取符合FIFO（先进先出）的顺序。队列头是在队列中呆的时间最长的元素，队尾是最短时间内排队的元素。

初始化、使用 `ArrayBlockingQueue`：

BlockingQueue queue = new ArrayBlockingQueue(1024); queue.put("1"); Object object = queue.take();

一个范型使用例子：

BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024); queue.put("1"); String string = queue.take();

[20180919155809666]: /images/20220612/23eaa21c66714f4a9ad373073037336c.png