聊聊Java中的生产者消费者模型——BlockingQueue
前言
生产者/消费者模型相信各位都不陌生，是一种很常见的分布式资源调度模型。在这个模型中，至少有两个对象：生产者和消费者。生产者只负责建立资源，消费者只负责使用资源。若是本身实现一个简单的生产者/消费者模型也很容易，无非就是经过一个队列来作，可是这种方式有不少隐藏的缺陷：java
- 须要保证资源的线程可见性，同时要手动实现线程同步
- 须要考虑各类临界状况和拒绝策略
- 须要在吞吐量与线程安全之间保持平衡
因此Java已经提早为咱们封装好了接口和实现，接下来咱们就要针对BlockingQueue接口和它的经常使用的实现类LinkedBlockingQueue进行简要的分析node
阻塞队列
概念
BlockingQueue，含义为阻塞队列，咱们能够从类定义看出，其继承了Queue接口，因此能够看成队列来使用：编程
既然叫作阻塞队列，也就是说这个队列的操做是以阻塞方式进行的，体如今以下两个方面：安全
- 插入元素的操做是阻塞的：当队列满时，执行插入操做的线程被阻塞
- 移除元素的操做时阻塞的：当队列空时，执行移除操做的线程被阻塞
经过这种方式，能够方便地协调生产者和消费者之间的关系并发
接口方法
在BlockingQueue中，定义了如下6个接口：分布式
public interface BlockingQueue<E> extends Queue<E> {
boolean add(E e);
boolean offer(E e);
void put(E e) throws InterruptedException;
boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
E take() throws InterruptedException;
E poll(long timeout, TimeUnit unit) throws InterruptedException;
int remainingCapacity();
boolean remove(Object o);
public boolean contains(Object o);
int drainTo(Collection<? super E> c);
int drainTo(Collection<? super E> c, int maxElements);
}
复制代码
这些接口方法按功能能够分为三类：post
- 添加元素：包括add、offer、put
- 移除元素：包括remove、poll、take、drainTo
- 获取/检查元素：包括contains、remainingCapacity
通常地，咱们也将添加元素叫作put操做（即便使用的是offer方法而不是put方法），移除元素的叫作take操做this
对于前两类，能够按照异常处理方式再次分为如下几类：spa
- 抛出异常：add、remove
- 返回特殊值：offer(e)、poll
- 阻塞：put(e)、take
- 超时退出：offer(e, time, unit)、poll(time, unit)
这几种处理方式我就很少解释了，字面意义已经很显然了线程
阻塞队列的实现
JDK8提供了如下BlockQueue的实现类：
咱们经常使用的基本有如下几种：
- ArrayBlockingQueue：基于ArrayList实现的阻塞队列，有界
- LinkedBlockingQueue：基于LinkedList实现的阻塞队列，有界
- PriorityBlockingQueue：优先队列，无界
- DelayQueue：支持延时获取元素的优先队列，无界
其他的实现感兴趣的能够自行了解，咱们这里就以LinkedBlockingQueue为例，介绍一下Java是如何实现阻塞队列的
接口方法
除了BlockingQueue提供的接口方法以外，LinkedBlockingQueue还提供了一个方法peek，用于获取队首节点
至此，咱们经常使用的阻塞队列方法都已说明完毕，这里用一张表来总结一下[1]：
| 方法/处理方式 | 抛出异常 | 返回特殊值 | 阻塞 | 超时退出 |
|---|---|---|---|---|
| 插入元素 | add(e) | offer(e) | put(e) | offer(e, timeout, unit) |
| 移除元素 | remove() | poll() | take() | poll(timeout, unit) |
| 获取元素 | element() | peek() | / | / |
其中element方法和peek方法功能是相同的
属性
BlockingQueue仅仅定义了接口规范，真正的实现仍是由具体的实现类来完成，咱们暂且略过中间的AbstractQueue，直接来研究LinkedBlockingQueue，其中定义了几个重要的域对象：
/** 元素个数 */
private final AtomicInteger count = new AtomicInteger();
/** 队首节点 */
transient Node<E> head;
/** 队尾节点 */
private transient Node<E> last;
/** take、poll等方法持有的锁，这里叫作take锁或出锁 */
private final ReentrantLock takeLock = new ReentrantLock();
/** take方法的等待队列 */
private final Condition notEmpty = takeLock.newCondition();
/** put、offer等方法持有的锁，这里叫作put锁或入锁 */
private final ReentrantLock putLock = new ReentrantLock();
/** put方法的等待队列 */
private final Condition notFull = putLock.newCondition();
复制代码
Node节点就是普通的队列节点，和LinkedList同样，咱们主要关注后面的4个域对象，能够分为两类：用于插入元素的，和用于移除元素的。其中每类都有两个属性：ReentranLock和Condition。其中ReentranLock是基于AQS[2]实现的一个可重入锁（不理解可重入概念的能够看成普通的锁），Condition是一个等待/通知模式的具体实现（能够理解为一种提供了功能更强大的wait和notify的类）
count属性天然不用说，head和last很显然是用于维护存储元素的队列，相信也不用细说了。阻塞队列和普通队列的区分点是在于后面的ReentrantLock和Condition类型的四个属性，关于这四个属性的意义，在接下来的几个模块会进行深刻的分析
不过咱们为了接下来说解方便，先来简单介绍一下Condition这个类。实际上，Condition是一个接口，具体的实现类是在AQS中。对于本篇文章来讲，你只须要清楚3个方法：await()、signal()，还有singalAll()。这三个方法彻底就能够类比wait()、notify()和notifyAll()，它们之间的区别能够模糊地理解为，wait/notify这些方法管理的是对象锁和类锁，它们操控的是等待这些锁的线程队列；而await/signal这些方法管理的是基于AQS的锁，操控的天然也是AQS中的线程等待队列
因此这里的notEmpty维护了等待take锁的线程队列，notFull维护了等待put锁的线程队列。从字面意义上也很好理解，notEmpty表示“队列还没空”，因此能够取元素，同理，notFull就表示“队列还没满”，能够往里插入元素
插入元素
offer(e)
先来看offer(e)方法，源码以下：
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
// 若是容量达到上限会返回false
if (count.get() == capacity)
return false;
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
// 获取put锁
putLock.lock();
try {
if (count.get() < capacity) {
// 入队并自增元素个数
enqueue(node);
// 注意，这里c返回的是增长前的值
c = count.getAndIncrement();
// 若是容量没到上限，就唤醒一个put操做
if (c + 1 < capacity)
notFull.signal();
}
} finally {
// 解锁
putLock.unlock();
}
if (c == 0)
// 若是队列以前为空，会唤醒一个take操做
signalNotEmpty();
return c >= 0;
}
复制代码
这个方法大部分操做都很好理解，当添加元素的操做不容许时，offer方法会给用户返回false，相似于非阻塞的通讯方式。offer方法的线程安全性是经过put锁来保证的
这里有一个颇有意思的地方，咱们看最后判断若是c == 0，那么就会唤醒一个take操做。可能不少人疑惑这里为何要加一条判断，是这样的，整个方法中，c的初值是-1，修改其值的惟一地方就是c = count.getAndIncrement()这条语句。也就是说，若是断定c == 0，那么这条语句的返回值就是0，即在插入元素以前，队列是空的。因此，若是一开始队列为空，当插入第一个元素以后，会马上唤醒一个take操做[3]
至此，整个方法流程能够概括为：
- 获取
put锁 - 元素入队，并自增
count值 - 若是容量未到上限，则唤醒一个
put操做 - 若是在插入元素以前队列为空，则在最后唤醒一个
take操做
offer(e, timeout, unit)
趁热打铁，咱们接着来看带有超时机制的offer方法：
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
// 可被中断地获取put锁
putLock.lockInterruptibly();
try {
// 重复执行while循环体，直到队列不满，或到了超时时间
while (count.get() == capacity) {
// 到了超时时间后就返回false
if (nanos <= 0)
return false;
// 会将当前线程添加到notFull等待队列中，
// 返回的是剩余可用的等待时间
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return true;
}
复制代码
整个方法大致上和offer(e)方法相同，不一样的点有两处：
- 获取锁采用的是可中断的形式，即
putLock.lockInterruptibly() - 若是队列一直是满的，则会循环执行
notFull.awaitNanos(nanos)操做来将当前线程添加到notFull等待队列中（等待put操做的执行）
其他部分和offer(e)彻底一致，在这里就不赘述了
add(e)
add方法与offer方法相比，当操做不容许时，会抛出异常而不是返回一个特殊值，以下：
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
复制代码
单纯地就是对offer(e)作了二次封装，没什么好说的，须要提一点的就是这个方法的实现是在AbstractQueue中
put(e)
put(e)方法当操做不容许时会阻塞线程，咱们来看其是如何实现的：
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
// 以可中断的形式获取put锁
putLock.lockInterruptibly();
try {
// 与offer(e, timeout, unit)相比，采用了无限等待的方式
while (count.get() == capacity) {
// 当执行了移除元素操做后，会经过signal操做来唤醒notFull队列中的一个线程
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
复制代码
果真方法之间都是大同小异的，put(e)操做能够类比咱们以前讲的offer(e, timeout, unit)，只有一个不一样的地方，就是当队列满时，await操做再也不有超时时间，也就是说，只能等待take操做[4]来调用signal方法唤醒该线程
移除元素
poll()
poll()方法用于移除并返回队首节点，下面是方法的具体实现：
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
// 获取take锁
takeLock.lock();
try {
if (count.get() > 0) {
// 出队，并自减
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
// 只要队列还有元素，就唤醒一个take操做
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
// 若是在队列满的状况下移除一个元素，会唤醒一个put操做
if (c == capacity)
signalNotFull();
return x;
}
复制代码
若是你认真看了offer(e)方法以后，poll()方法就没什么好讲的了，彻底就是offer(e)的翻版（我也想讲点东西，可是poll()方法彻底和offer(e)流程如出一辙...）
其余
poll(timeout, unit)/take()/remove()方法分别是offer(e, timeout, unit)/put()/add()方法的翻版，没有什么特殊的地方，这里就一笔略过了
获取元素
peek()
peek()方法是用于获取队首元素，其实现以下：
public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
// 获取take锁
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}
复制代码
流程没什么好说的，须要注意的是该方法须要获取take锁，也就是说在peek()方法执行时，是不能执行移除元素的操做的
element()
element()方法的实现是在AbstractQueue中：
public E element() {
E x = peek();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
复制代码
仍是一样的二次封装操做
总结
原本说的是BlockingQueue，结果说了半天LinkedBlockingQueue。不过做为阻塞队列的一种经典实现，LinkedBlockingQueue中的方法实现思路也是对于理解阻塞队列来讲也是很重要的。想要理解阻塞队列的理念，最重要的就是理解锁的概念，好比LinkedBlockingQueue经过生产者锁/put锁和消费者锁/take锁，以及锁对应的Condition对象来实现线程安全。理解了这一点，才能理解整个生产者/消费者模型
这里参考了《Java并发编程的艺术》 ↩︎
参见浅谈AQS（抽象队列同步器）一文 ↩︎
这里描述为“唤醒一个
take操做”有些不许确，实际应描述为“唤醒一个等待take锁的线程”，不过我认为前者更有助于读者理解，因此沿用前者的描述方式 ↩︎指的是和
take功能相似的一组方法，包含有take/poll/remove，put操做同理 ↩︎
- 1. Java 生产者消费者实现——BlockingQueue
- 2. Java多线程（十）：BlockingQueue实现生产者消费者模型
- 3. 聊聊并发（十）生产者消费者模式
- 4. Java--生产者消费者模型
- 5. 生产者与消费者（三）---BlockingQueue
- 6. 生产消费者模型
- 7. 生产者消费模型
- 8. day50-1 生产者消费者模型
- 9. 生产者消费者模型
- 10. 生产者与消费者模型
- 更多相关文章...
- • 现实生活中的 XML - XML 教程
- • Java 中操作 R - R 语言教程
- • 适用于PHP初学者的学习线路和建议
- • 委托模式
-
每一个你不满意的现在，都有一个你没有努力的曾经。