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》