生产者消费者模式经常在面试中问到,并且很多面试官或者力扣的算法题中明确要求自己实现阻塞队列。
什么是生产者消费者模式
生产者消费者模式是程序设计中非常常见的一种设计模式,被广泛运用在解耦、消息队列等场景。在现实世界中,我们把生产商品的一方称为生产者,把消费商品的一方称为消费者,有时生产者的生产速度特别快,但消费者的消费速度跟不上,俗称“产能过剩”,又或是多个生产者对应多个消费者时,大家可能会手忙脚乱。如何才能让大家更好地配合呢?这时在生产者和消费者之间就需要一个中介来进行调度,于是便诞生了生产者消费者模式。
其实就是MQ。用来做缓冲作用。如下图所示,右侧的1是生产者线程,生产者在生产数据后将数据存放在阻塞队列中,左侧的2是消费者线程,消费者获取阻塞队列中的数据。而中间的3和4分别代表生产者消费者之前互相通信的过程,因为无论是阻塞队列是满还是空都将阻塞,阻塞之后就会在满足条件时去唤醒被阻塞的线程。
使用BlockingQueue实现生产者消费者模式
public static void main(String[] args) {
BlockingQueue<Object> queue = new ArrayBlockingQueue<>(10);
Runnable producer = () -> {
while (true) {
queue.put(new Object());
}
};
new Thread(producer).start();
new Thread(producer).start();
Runnable consumer = () -> {
while (true) {
queue.take();
}
};
new Thread(consumer).start();
new Thread(consumer).start();
}
以上代码通过创建一个ArrayBlockingQueue类型的BlockingQueue,并初始化队列大小为10。然后创建2个生产者线程负责往队列中添加数据;创建两个消费者线程负责消费数据。
其中的put和take方法中,通过ReentrantLock做了一些同步策略。
使用Condition实现生产者消费者模式
使用Condition来实现生产者消费者模式,实质是模拟BlockingQueue的实现方式。代码如下。
public class MyBlockingQueueForCondition {
private Queue queue;
private int max = 16;
private ReentrantLock lock = new ReentrantLock();
private Condition notEmpty = lock.newCondition();
private Condition notFull = lock.newCondition();
public MyBlockingQueueForCondition(int size) {
this.max = size;
queue = new LinkedList();
}
public void put(Object o) throws InterruptedException {
lock.lock();
try {
while (queue.size() == max) {
notFull.await();
}
queue.add(o);
notEmpty.signalAll();
} finally {
lock.unlock();
}
}
public Object take() throws InterruptedException {
lock.lock();
try {
while (queue.size() == 0) {
notEmpty.await();
}
Object item = queue.remove();
notFull.signalAll();
return item;
} finally {
lock.unlock();
}
}
}
这次是直接使用java中的Queue对象,并初始化大小为16,其次定义了一个ReentrantLock类型的Lock锁,并在Lock锁的基础上创建两个Condition,一个是notEmpty,另一个是notFull,分别表示队列没有空和没有满的条件;最后,声明了put和take这两个核心方法。
这里需要自己实现同步措施保证线程安全,所以在put方法中先将Lock锁上,然后,在while的条件里检测queue是不是已经满了,如果已经满了,则调用notFull的await()阻塞生产者线程并释放Lock,如果没有满,则往队列放入数据并利用notEmpty.signalAll()通知正在等待的所有消费者并唤醒它们。最后在finally中利用lock.unlock()方法解锁,** 把unlock方法放在finally中是一个基本原则,否则可能会产生无法释放的情况。**
下面再来看 take 方法,take 方法实际上是与 put 方法相互对应的,同样是通过 while 检查队列是否为空,如果为空,消费者开始等待,如果不为空则从队列中获取数据并通知生产者队列有空余位置,最后在 finally 中解锁。
这里需要注意,我们在 take() 方法中使用 while( queue.size() == 0 )
检查队列状态,而不能用 if( queue.size() == 0 )
。为什么呢?大家思考这样一种情况,因为生产者消费者往往是多线程的,我们假设有两个消费者,第一个消费者线程获取数据时,发现队列为空,便进入等待状态;因为第一个线程在等待时会释放 Lock 锁,所以第二个消费者可以进入并执行 if( queue.size() == 0 )
,也发现队列为空,于是第二个线程也进入等待;而此时,如果生产者生产了一个数据,便会唤醒两个消费者线程,而两个线程中只有一个线程可以拿到锁,并执行 queue.remove 操作,另外一个线程因为没有拿到锁而卡在被唤醒的地方,而第一个线程执行完操作后会在 finally 中通过 unlock 解锁,而此时第二个线程便可以拿到被第一个线程释放的锁,继续执行操作,也会去调用 queue.remove 操作,然而这个时候队列已经为空了,所以会抛出 NoSuchElementException 异常,这不符合我们的逻辑。而如果用 while 做检查,当第一个消费者被唤醒得到锁并移除数据之后,第二个线程在执行 remove 前仍会进行 while 检查,发现此时依然满足 queue.size() == 0 的条件,就会继续执行 await 方法,避免了获取的数据为 null 或抛出异常的情况。
使用wait/notify结合链表实现生产者消费者模式
class MyBlockingQueue {
private int maxSize;
private LinkedList<Object> storage;
public MyBlockingQueue(int size) {
this.maxSize = size;
storage = new LinkedList<>();
}
public synchronized void put() throws InterruptedException {
while (storage.size() == maxSize) {
wait();
}
storage.add(new Object());
notifyAll();
}
public synchronized void take() throws InterruptedException {
while (storage.size() == 0) {
wait();
}
System.out.println(storage.remove());
notifyAll();
}
}
如代码所示,最主要的部分仍是 take 与 put 方法,我们先来看 put 方法,put 方法被 synchronized 保护,while 检查队列是否为满,如果不满就往里放入数据并通过 notifyAll() 唤醒其他线程。同样,take 方法也被 synchronized 修饰,while 检查队列是否为空,如果不为空就获取数据并唤醒其他线程。使用这个 MyBlockingQueue 实现的生产者消费者代码如下:
/**
* 描述: wait形式实现生产者消费者模式
*/
public class WaitStyle {
public static void main(String[] args) {
MyBlockingQueue myBlockingQueue = new MyBlockingQueue(10);
Producer producer = new Producer(myBlockingQueue);
Consumer consumer = new Consumer(myBlockingQueue);
new Thread(producer).start();
new Thread(consumer).start();
}
}
class Producer implements Runnable {
private MyBlockingQueue storage;
public Producer(MyBlockingQueue storage) {
this.storage = storage;
}
@Override
public void run() {
for (int i = 0; i < 100; i++) {
try {
storage.put();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class Consumer implements Runnable {
private MyBlockingQueue storage;
public Consumer(MyBlockingQueue storage) {
this.storage = storage;
}
@Override
public void run() {
for (int i = 0; i < 100; i++) {
try {
storage.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
以上就是实现生产者消费者的三种方式。本质上都是实现一个阻塞队列来达到生产者消费者功能的目的。