Java里的生产者-消费者模型(Producer and Consumer Pattern in Java)

生产者-消费者模型是多线程问题里面的经典问题,也是面试的常见问题。有如下几个常见的实现方法:

1. wait()/notify()

2. lock & condition

3. BlockingQueue

下面来逐一分析。

1. wait()/notify()

第一种实现,利用根类Object的两个方法wait()/notify(),来停止或者唤醒线程的执行;这也是最原始的实现。

 public class WaitNotifyBroker<T> implements Broker<T> {

     private final Object[] items;

     private int takeIndex;
private int putIndex;
private int count; public WaitNotifyBroker(int capacity) {
this.items = new Object[capacity];
} @SuppressWarnings("unchecked")
@Override
public T take() {
T tmpObj = null;
try {
synchronized (items) {
while (0 == count) {
items.wait();
}
tmpObj = (T) items[takeIndex];
if (++takeIndex == items.length) {
takeIndex = 0;
}
count--;
items.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
} return tmpObj;
} @Override
public void put(T obj) {
try {
synchronized (items) {
while (items.length == count) {
items.wait();
} items[putIndex] = obj;
if (++putIndex == items.length) {
putIndex = 0;
}
count++;
items.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
} } }

这里利用Array构造一个Buffer去存取数据,并利用count, putIndex和takeIndex来保证First-In-First-Out。

如果利用LinkedList来代替Array,相对来说会稍微简单些。

LinkedList的实现,可以参考《Java 7 Concurrency Cookbook》第2章wait/notify。

2. lock & condition

lock & condition,实际上也实现了类似synchronized和wait()/notify()的功能,但在加锁和解锁、暂停和唤醒方面,更加细腻和可控。

在JDK的BlockingQueue的默认实现里,也是利用了lock & condition。此文也详细介绍了怎么利用lock&condition写BlockingQueue,这里换LinkedList再实现一次:

 public class LockConditionBroker<T> implements Broker<T> {

     private final ReentrantLock lock;
private final Condition notFull;
private final Condition notEmpty;
private final int capacity;
private LinkedList<T> items; public LockConditionBroker(int capacity) {
this.lock = new ReentrantLock();
this.notFull = lock.newCondition();
this.notEmpty = lock.newCondition();
this.capacity = capacity; items = new LinkedList<T>();
} @Override
public T take() {
T tmpObj = null;
lock.lock();
try {
while (items.size() == 0) {
notEmpty.await();
} tmpObj = items.poll();
notFull.signalAll(); } catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return tmpObj;
} @Override
public void put(T obj) {
lock.lock();
try {
while (items.size() == capacity) {
notFull.await();
} items.offer(obj);
notEmpty.signalAll(); } catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
} }
}

3. BlockingQueue

最后这种方法,也是最简单最值得推荐的。利用并发包提供的工具:阻塞队列,将阻塞的逻辑交给BlockingQueue。

实际上,上述1和2的方法实现的Broker类,也可以视为一种简单的阻塞队列,不过没有标准包那么完善。

 public class BlockingQueueBroker<T> implements Broker<T> {

     private final BlockingQueue<T> queue;

     public BlockingQueueBroker() {
this.queue = new LinkedBlockingQueue<T>();
} @Override
public T take() {
try {
return queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
} return null;
} @Override
public void put(T obj) {
try {
queue.put(obj);
} catch (InterruptedException e) {
e.printStackTrace();
}
} }

我们的队列封装了标注包里的LinkedBlockingQueue,十分简单高效。

接下来,就是一个1P2C的例子:

 public interface Broker<T> {

     T take();

     void put(T obj);

 }

 public class Producer implements Runnable {

     private final Broker<Integer> broker;
private final String name; public Producer(Broker<Integer> broker, String name) {
this.broker = broker;
this.name = name;
} @Override
public void run() {
try {
for (int i = 0; i < 5; i++) {
broker.put(i);
System.out.format("%s produced: %s%n", name, i);
Thread.sleep(1000);
}
broker.put(-1);
System.out.println("produced termination signal");
} catch (InterruptedException e) {
e.printStackTrace();
return;
} } } public class Consumer implements Runnable { private final Broker<Integer> broker;
private final String name; public Consumer(Broker<Integer> broker, String name) {
this.broker = broker;
this.name = name;
} @Override
public void run() {
try {
for (Integer message = broker.take(); message != -1; message = broker.take()) {
System.out.format("%s consumed: %s%n", name, message);
Thread.sleep(1000);
}
System.out.println("received termination signal");
} catch (InterruptedException e) {
e.printStackTrace();
return;
} } } public class Main { public static void main(String[] args) {
Broker<Integer> broker = new WaitNotifyBroker<Integer>(5);
// Broker<Integer> broker = new LockConditionBroker<Integer>(5);
// Broker<Integer> broker = new BlockingQueueBroker<Integer>(); new Thread(new Producer(broker, "prod 1")).start();
new Thread(new Consumer(broker, "cons 1")).start();
new Thread(new Consumer(broker, "cons 2")).start(); } }

除了上述的方法,其实还有很多第三方的并发包可以解决这个问题。例如LMAX Disruptor和Chronicle等

本文完。

参考:

《Java 7 Concurrency Cookbook》

上一篇:Java实现多线程生产者消费者模型及优化方案


下一篇:mysql中explain输出列之id的用法详解