[Java] BlockingQueue 简介

拼搏现实的明天。 2022-09-30 03:58 121阅读 0赞

`BlockingQueue` 接口表示它是一个 `Queue`，意思是存储在Queue 中的元素以先入先出（FIFO）顺序存储。在特定顺序插入的元素以相同的顺序检索 — 但是需要附加保证，从空队列检索一个元素的操作都会阻塞调用线程，直到这个项准备好被检索。同理，想要将一个元素插入到满队列的尝试也会导致阻塞调用线程，直到队列的存储空间可用。

`BlockingQueue` 干净利落地解决了如何将一个线程收集的项“传递”给另一线程用于处理的问题，无需考虑同步问题。

BlockingQueue 支持四种类型的操作。如下：

<table> 
 <tbody> 
  <tr> 
   <td>&nbsp;</td> 
   <td align="center"><em>Throws exception</em></td> 
   <td align="center"><em>Special value</em></td> 
   <td align="center"><em>Blocks</em></td> 
   <td align="center"><em>Times out</em></td> 
  </tr> 
  <tr> 
   <td><strong>Insert</strong></td> 
   <td><code>add(e)</code></td> 
   <td><code>offer(e)</code></td> 
   <td><code>put(e)</code></td> 
   <td><code>offer(e, time, unit)</code></td> 
  </tr> 
  <tr> 
   <td><strong>Remove</strong></td> 
   <td><code>remove()</code></td> 
   <td><code>poll()</code></td> 
   <td><code>take()</code></td> 
   <td><code>poll(time, unit)</code></td> 
  </tr> 
  <tr> 
   <td><strong>Examine</strong></td> 
   <td><code>element()</code></td> 
   <td><code>peek()</code></td> 
   <td><em>not applicable</em></td> 
   <td><em>not applicable</em></td> 
  </tr> 
 </tbody> 
</table>

1 add(e) remove() element() 方法不会阻塞线程，当不满足约束条件时，会抛出IllegalStateException 异常。例如：当queue 被元素填满后，在调用add(e)，则 该方法会抛出异常。

2 offer(e) poll() peek() 方法也不会阻塞线程，也会抛出异常。例如：当queue 被元素填满后，在调用offer(e)，则不会插入元素，函数返回false。

3 想要实现阻塞功能，需要调用put(e) take() 方法。当不满足约束条件时，会阻塞线程。

下图为类的继承情况：其中ArrayBlockingQueue 和 LinkedBlockingQueue 是两个常用的类。Deque 为双向列表，即头和尾都能插入和删除元素。这两个类都是线程安全的。

![0_1302231325nF1m.gif][]

我们在这里以ArrayBlockingQueue 为例，来看看其中的put 和take 方法的实现。

put 方法：

/\*\* \* Inserts the specified element at the tail of this queue, waiting \* for space to become available if the queue is full. \* \* @throws InterruptedException \{@inheritDoc\} \* @throws NullPointerException \{@inheritDoc\} \*/ public void put(E e) throws InterruptedException \{ if (e == null) throw new NullPointerException(); final E\[\] items = this.items; final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try \{ try \{ while (count == items.length) notFull.await(); \} catch (InterruptedException ie) \{ notFull.signal(); // propagate to non-interrupted thread throw ie; \} insert(e); \} finally \{ lock.unlock(); \} \}

可以看出，ArrayBlockingQueue 内部由一个数组来实现队列的，内部数组的大小在构造函数中指定。插入前使用lock.lockInterruptibly() 进行了加锁。如果数组满了，就会等待（notFull.await() 实现）；未满则插入。

take 方法：

public E take() throws InterruptedException \{ final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try \{ try \{ while (count == 0) notEmpty.await(); \} catch (InterruptedException ie) \{ notEmpty.signal(); // propagate to non-interrupted thread throw ie; \} E x = extract(); return x; \} finally \{ lock.unlock(); \} \}

实现方法和 put 大同小异。

用法实例：

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(); \} \}

[0_1302231325nF1m.gif]: /images/20220708/c1ca600cd96c4198998a246f63f8fb32.png