【Java数据结构及算法实战】系列008：Java队列02——阻塞队列BlockingQueue

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

阻塞队列（BlockingQueue）是一种支持额外操作的队列，这两个附加的操作是：

l 在队列为空时，获取元素的线程会等待队列变为非空。

l 当队列满时，存储元素的线程会等待队列可用。

Java提供了java.util.concurrent.BlockingQueue<E>接口以提供对阻塞队列的支持。该接口是Java Collections Framework的一个成员。

1. BlockingQueue的方法

BlockingQueue对不能立即满足的操作有不同的处理方法，这些方法共有四种形式：

l 抛出一个异常

l 返回一个特殊值，根据操作的不同可能返回null或false

l 无限期地阻塞当前线程，直到操作成功

l 在放弃之前只阻塞给定的最大时间限制。

这些方法总结如下：

方法\处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e) 队列未满时，返回 true；队列满则抛出异常	offer(e) 队列未满时，返回 true；队列满时返回 false。非阻塞立即返回。	put(e) 队列未满时，直接插入没有返回值；队列满时会阻塞等待，一直等到队列未满时再插入。	offer(e,time,unit) 设定等待的时间，如果在指定时间内还不能往队列中插入数据则返回 false。插入成功返回 true。
移除方法	remove() 队列不为空时，返回队首值并移除队列为空时抛出异常	poll() 队列不为空时返回队首值并移除队列为空时返回 null。非阻塞立即返回。	take() 队列不为空返回队首值并移除当队列为空时会阻塞等待，一直等到队列不为空时再返回队首值。	poll(time,unit) 设定等待的时间，如果在指定时间内队列还未孔则返回 null 不为空则返回队首值
检查方法	element() 队列不为空时返回队首值但不移除队列为空时抛出异常。	peek() 队列不为空时返回队首值但不移除队列为空时返回 null。	不可用	不可用

BlockingQueue不接受null值，当尝试add、put或offer一个null值时，将抛出NullPointerException。null值有特殊的用意，用来表示poll操作失败。

BlockingQueue可以有容量限制，如果不指定容量则默认容量大小是Integer.MAX_VALUE。任何时候，只要超过了剩余容量（remainingCapacity），则新的元素不能被放入队列中。

BlockingQueue实现被设计为主要用于生产者-消费者队列，但还支持Collection接口。因此，例如，使用remove(x)可以从队列中删除任意元素。然而，这些操作通常不会非常有效地执行，并且只用于偶尔使用，例如在取消排队的消息的时候。

BlockingQueue实现是线程安全的。所有排队方法都使用内部锁或其他并发控制形式自动地实现其效果。但是，除非在实现中指定了otherwise，否则批量Collection操作addAll、containsAll、retainAll和removeAll不一定是自动执行的。因此，例如，在只添加c中的某些元素之后， addAll ( c )可能会失败（抛出一个异常）。

BlockingQueue本质上不支持任何类型的close或shutdown操作，以指示不再添加任何项。这些特性的需求和使用往往依赖于具体的实现。例如，一种常见的策略是生产者插入特殊的对象（比如end-of-stream或 poison），当被消费者采用时，这些对象将得到相应的解释。

2. BlockingQueue的使用示例

以下是一个典型的生产者-消费者场景的使用示例。注意，一个BlockingQueue可以安全地与多个生产者和多个消费者一起使用。

class Producer implements Runnable {
   private final BlockingQueue queue;
   Producer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while (true) { queue.put(produce()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   Object produce() { ... }
 }
 class Consumer implements Runnable {
   private final BlockingQueue queue;
   Consumer(BlockingQueue q) { queue = q; }
   public void run() {
     try {
       while (true) { consume(queue.take()); }
     } catch (InterruptedException ex) { ... handle ...}
   }
   void consume(Object x) { ... }
 }
 class Setup {
   void main() {
     BlockingQueue q = new SomeQueueImplementation();
     Producer p = new Producer(q);
     Consumer c1 = new Consumer(q);
     Consumer c2 = new Consumer(q);
     new Thread(p).start();
     new Thread(c1).start();
     new Thread(c2).start();
   }
 }}