生產(chǎn)者消費(fèi)者模型

生產(chǎn)者消費(fèi)者模型可以描述為：
①生產(chǎn)者持續(xù)生產(chǎn)，直到倉庫放滿產(chǎn)品，則停止生產(chǎn)進(jìn)入等待狀態(tài)；倉庫不滿后繼續(xù)生產(chǎn)；
②消費(fèi)者持續(xù)消費(fèi)，直到倉庫空，則停止消費(fèi)進(jìn)入等待狀態(tài)；倉庫不空后，繼續(xù)消費(fèi)；
③生產(chǎn)者可以有多個，消費(fèi)者也可以有多個；

生產(chǎn)者消費(fèi)者模型

對應(yīng)到程序中，倉庫對應(yīng)緩沖區(qū)，可以使用隊(duì)列來作為緩沖區(qū)，并且這個隊(duì)列應(yīng)該是有界的，即最大容量是固定的；進(jìn)入等待狀態(tài)，則表示要阻塞當(dāng)前線程，直到某一條件滿足，再進(jìn)行喚醒。

常見的實(shí)現(xiàn)方式主要有以下幾種。
①使用wait()和notify()
②使用Lock和Condition
③使用信號量Semaphore
④使用JDK自帶的阻塞隊(duì)列
⑤使用管道流

使用wait()和notify()實(shí)現(xiàn)

前提是要熟悉Object的幾個方法：

• wait():當(dāng)前線程釋放鎖，直到等到通知，再去獲取鎖
• sleep():當(dāng)前線程休眠，但不釋放鎖
• notify():喚醒其他正在wait的線程

參考代碼如下：

public class ProducerConsumer1 {

    class Producer extends Thread {
        private String threadName;
        private Queue<Goods> queue;
        private int maxSize;

        public Producer(String threadName, Queue<Goods> queue, int maxSize) {
            this.threadName = threadName;
            this.queue = queue;
            this.maxSize = maxSize;
        }

        @Override
        public void run() {
            while (true) {
                //模擬生產(chǎn)過程中的耗時(shí)操作
                Goods goods = new Goods();
                try {
                    Thread.sleep(new Random().nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                synchronized (queue) {
                    while (queue.size() == maxSize) {
                        try {
                            System.out.println("隊(duì)列已滿，【" + threadName + "】進(jìn)入等待狀態(tài)");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    queue.add(goods);
                    System.out.println("【" + threadName + "】生產(chǎn)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                    queue.notifyAll();
                }
            }
        }
    }

    class Consumer extends Thread {
        private String threadName;
        private Queue<Goods> queue;

        public Consumer(String threadName, Queue<Goods> queue) {
            this.threadName = threadName;
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true) {
                Goods goods;
                synchronized (queue) {
                    while (queue.isEmpty()) {
                        try {
                            System.out.println("隊(duì)列已空，【" + threadName + "】進(jìn)入等待狀態(tài)");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    goods = queue.remove();
                    System.out.println("【" + threadName + "】消費(fèi)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                    queue.notifyAll();
                }
                //模擬消費(fèi)過程中的耗時(shí)操作
                try {
                    Thread.sleep(new Random().nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    @Test
    public void test() {

        int maxSize = 5;
        Queue<Goods> queue = new LinkedList<>();

        Thread producer1 = new Producer("生產(chǎn)者1", queue, maxSize);
        Thread producer2 = new Producer("生產(chǎn)者2", queue, maxSize);
        Thread producer3 = new Producer("生產(chǎn)者3", queue, maxSize);

        Thread consumer1 = new Consumer("消費(fèi)者1", queue);
        Thread consumer2 = new Consumer("消費(fèi)者2", queue);


        producer1.start();
        producer2.start();
        producer3.start();
        consumer1.start();
        consumer2.start();

        while (true) {

        }
    }
}

幾個注意的地方：

①確定鎖的對象是隊(duì)列queue；

②不要把生產(chǎn)過程和消費(fèi)過程寫在同步塊中，這些操作無需同步，同步的僅僅是放入和取出這兩個動作；

③因?yàn)槭浅掷m(xù)生產(chǎn)，持續(xù)消費(fèi)，要用while(true){...}的方式將【生產(chǎn)、放入】或【取出、消費(fèi)】的操作都一直進(jìn)行。

④但由于是對隊(duì)列使用synchronized的方式加鎖，同一時(shí)刻，要么在放入，要么在取出，兩者不能同時(shí)進(jìn)行。

使用Lock和Condition實(shí)現(xiàn)

前提是要熟悉Lock接口以及常用實(shí)現(xiàn)類ReentrantLock，以及Condition的兩個常用方法：

• await():等待Condition的滿足，會釋放鎖
• signal():喚醒其他正在等待該Condition的線程
  參考代碼如下：

public class ProducerConsumer2 {

    class Producer extends Thread {

        private String threadName;
        private Queue<Goods> queue;
        private Lock lock;
        private Condition notFullCondition;
        private Condition notEmptyCondition;
        private int maxSize;

        public Producer(String threadName, Queue<Goods> queue, Lock lock, Condition notFullCondition, Condition notEmptyCondition, int maxSize) {
            this.threadName = threadName;
            this.queue = queue;
            this.lock = lock;
            this.notFullCondition = notFullCondition;
            this.notEmptyCondition = notEmptyCondition;
            this.maxSize = maxSize;

        }

        @Override
        public void run() {
            while (true) {
                //模擬生產(chǎn)過程中的耗時(shí)操作
                Goods goods = new Goods();
                try {
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                lock.lock();
                try {
                    while (queue.size() == maxSize) {
                        try {
                            System.out.println("隊(duì)列已滿，【" + threadName + "】進(jìn)入等待狀態(tài)");
                            notFullCondition.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    queue.add(goods);
                    System.out.println("【" + threadName + "】生產(chǎn)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                    notEmptyCondition.signalAll();

                } finally {
                    lock.unlock();
                }
            }
        }
    }

    class Consumer extends Thread {
        private String threadName;
        private Queue<Goods> queue;
        private Lock lock;
        private Condition notFullCondition;
        private Condition notEmptyCondition;

        public Consumer(String threadName, Queue<Goods> queue, Lock lock, Condition notFullCondition, Condition notEmptyCondition) {
            this.threadName = threadName;
            this.queue = queue;
            this.lock = lock;
            this.notFullCondition = notFullCondition;
            this.notEmptyCondition = notEmptyCondition;
        }

        @Override
        public void run() {
            while (true) {
                Goods goods;
                lock.lock();
                try {
                    while (queue.isEmpty()) {
                        try {
                            System.out.println("隊(duì)列已空，【" + threadName + "】進(jìn)入等待狀態(tài)");
                            notEmptyCondition.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    goods = queue.remove();
                    System.out.println("【" + threadName + "】消費(fèi)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                    notFullCondition.signalAll();

                } finally {
                    lock.unlock();
                }

                //模擬消費(fèi)過程中的耗時(shí)操作
                try {
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Test
    public void test() {

        int maxSize = 5;
        Queue<Goods> queue = new LinkedList<>();
        Lock lock = new ReentrantLock();
        Condition notEmptyCondition = lock.newCondition();
        Condition notFullCondition = lock.newCondition();

        Thread producer1 = new ProducerConsumer2.Producer("生產(chǎn)者1", queue, lock, notFullCondition, notEmptyCondition, maxSize);
        Thread producer2 = new ProducerConsumer2.Producer("生產(chǎn)者2", queue, lock, notFullCondition, notEmptyCondition, maxSize);
        Thread producer3 = new ProducerConsumer2.Producer("生產(chǎn)者3", queue, lock, notFullCondition, notEmptyCondition, maxSize);

        Thread consumer1 = new ProducerConsumer2.Consumer("消費(fèi)者1", queue, lock, notFullCondition, notEmptyCondition);
        Thread consumer2 = new ProducerConsumer2.Consumer("消費(fèi)者2", queue, lock, notFullCondition, notEmptyCondition);
        Thread consumer3 = new ProducerConsumer2.Consumer("消費(fèi)者3", queue, lock, notFullCondition, notEmptyCondition);


        producer1.start();
        producer2.start();
        producer3.start();
        consumer1.start();
        consumer2.start();
        consumer3.start();
        while (true) {

        }
    }
}

要注意的地方：

放入和取出操作均是用的同一個鎖，所以在同一時(shí)刻，要么在放入，要么在取出，兩者不能同時(shí)進(jìn)行。因此，與使用wait()和notify()實(shí)現(xiàn)類似，這種方式的實(shí)現(xiàn)并不能最大限度地利用緩沖區(qū)（即例子中的隊(duì)列）。如果要實(shí)現(xiàn)同一時(shí)刻，既可以放入又可以取出，則要使用兩個重入鎖，分別控制放入和取出的操作，具體實(shí)現(xiàn)可以參考LinkedBlockingQueue。

使用信號量Semaphore實(shí)現(xiàn)

前提是熟悉信號量Semaphore的使用方式，尤其是release()方法，Semaphore在release之前不必一定要先acquire。(如果不熟悉Semaphore，可以參考閱讀【多線程與并發(fā)】Java并發(fā)工具類)

There is no requirement that a thread that releases a permit must
have acquired that permit by calling acquire.
Correct usage of a semaphore is established by programming convention
in the application.

參考代碼如下：

public class ProducerConsumer4 {


    class Producer extends Thread {
        private String threadName;
        private Queue<Goods> queue;
        private Semaphore queueSizeSemaphore;
        private Semaphore concurrentWriteSemaphore;
        private Semaphore notEmptySemaphore;

        public Producer(String threadName, Queue<Goods> queue, Semaphore concurrentWriteSemaphore, Semaphore queueSizeSemaphore, Semaphore notEmptySemaphore) {
            this.threadName = threadName;
            this.queue = queue;
            this.concurrentWriteSemaphore = concurrentWriteSemaphore;
            this.queueSizeSemaphore = queueSizeSemaphore;
            this.notEmptySemaphore = notEmptySemaphore;
        }

        @Override
        public void run() {
            while (true) {
                //模擬生產(chǎn)過程中的耗時(shí)操作
                Goods goods = new Goods();
                try {
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                try {
                    queueSizeSemaphore.acquire();//獲取隊(duì)列未滿的信號量
                    concurrentWriteSemaphore.acquire();//獲取讀寫的信號量
                    queue.add(goods);
                    System.out.println("【" + threadName + "】生產(chǎn)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    concurrentWriteSemaphore.release();
                    notEmptySemaphore.release();
                }
            }
        }
    }

    class Consumer extends Thread {
        private String threadName;
        private Queue<Goods> queue;
        private Semaphore queueSizeSemaphore;
        private Semaphore concurrentWriteSemaphore;
        private Semaphore notEmptySemaphore;

        public Consumer(String threadName, Queue<Goods> queue, Semaphore concurrentWriteSemaphore, Semaphore queueSizeSemaphore, Semaphore notEmptySemaphore) {
            this.threadName = threadName;
            this.queue = queue;
            this.concurrentWriteSemaphore = concurrentWriteSemaphore;
            this.queueSizeSemaphore = queueSizeSemaphore;
            this.notEmptySemaphore = notEmptySemaphore;
        }

        @Override
        public void run() {
            while (true) {
                Goods goods;
                try {
                    notEmptySemaphore.acquire();
                    concurrentWriteSemaphore.acquire();
                    goods = queue.remove();
                    System.out.println("【" + threadName + "】生產(chǎn)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    concurrentWriteSemaphore.release();
                    queueSizeSemaphore.release();
                }

                //模擬消費(fèi)過程中的耗時(shí)操作
                try {
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    @Test
    public void test() {

        int maxSize = 5;
        Queue<Goods> queue = new LinkedList<>();
        Semaphore concurrentWriteSemaphore = new Semaphore(1);
        Semaphore notEmptySemaphore = new Semaphore(0);
        Semaphore queueSizeSemaphore = new Semaphore(maxSize);


        Thread producer1 = new ProducerConsumer4.Producer("生產(chǎn)者1", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);
        Thread producer2 = new ProducerConsumer4.Producer("生產(chǎn)者2", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);
        Thread producer3 = new ProducerConsumer4.Producer("生產(chǎn)者3", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);

        Thread consumer1 = new ProducerConsumer4.Consumer("消費(fèi)者1", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);
        Thread consumer2 = new ProducerConsumer4.Consumer("消費(fèi)者2", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);
        Thread consumer3 = new ProducerConsumer4.Consumer("消費(fèi)者3", queue, concurrentWriteSemaphore, queueSizeSemaphore, notEmptySemaphore);


        producer1.start();
        producer2.start();
        producer3.start();
        consumer1.start();
        consumer2.start();
        consumer3.start();
        while (true) {
        }
    }
}

要注意的地方：

①理解代碼中的三個信號量的含義
queueSizeSemaphore：（其中的許可證數(shù)量，可以理解為隊(duì)列中可以再放入多少個元素），該信號量的許可證初始數(shù)量為倉庫大小，即maxSize；生產(chǎn)者每放置一個商品，則該信號量-1，即執(zhí)行acquire()，表示隊(duì)列中已經(jīng)添加了一個元素，要減少一個許可證；消費(fèi)者每取出一個商品，該信號量+1，即執(zhí)行release()，表示隊(duì)列中已經(jīng)少了一個元素，再給你一個許可證。
notEmptySemaphore：（其中的許可證數(shù)量，可以理解為隊(duì)列中可以取出多少個元素），該信號量的許可證初始數(shù)量為0；生產(chǎn)者每放置一個商品，則該信號量+1，即執(zhí)行release()，表示隊(duì)列中添加了一個元素；消費(fèi)者每取出一個商品，該信號量-1，即執(zhí)行acquire()，表示隊(duì)列中已經(jīng)少了一個元素，要減少一個許可證；
concurrentWriteSemaphore，相當(dāng)于一個寫鎖，在放入或取出商品的時(shí)候，都需要先獲取再釋放許可證。

②由于實(shí)現(xiàn)中，使用了concurrentWriteSemaphore實(shí)現(xiàn)了對隊(duì)列并發(fā)寫的控制，在同一時(shí)刻，只能對隊(duì)列進(jìn)行一種操作：放入或取出。假如把concurrentWriteSemaphore中的信號量初始化為2或者2以上的值，就會出現(xiàn)多個生產(chǎn)者同時(shí)放入或多個消費(fèi)者同時(shí)消費(fèi)的情況，而使用的LinkedList是不允許并發(fā)進(jìn)行這種修改的，否則會出現(xiàn)溢出或取空的情況。所以，concurrentWriteSemaphore只能設(shè)置為1，也就導(dǎo)致性能與使用wait() / notify()方式類似，性能不高。

使用jdk自帶的阻塞隊(duì)列實(shí)現(xiàn)

前提是要記住兩個阻塞取放方法，因?yàn)樽枞?duì)列提供了很多存取元素的方法，幾種存取方式在隊(duì)列已滿/已空時(shí)采取的措施如下：

所以，在這里，要選用put()和take()這兩個會阻塞的方法。

參考代碼如下：

public class ProducerConsumer3 {

    class Producer extends Thread {
        private String threadName;
        private BlockingQueue<Goods> queue;

        public Producer(String threadName, BlockingQueue<Goods> queue) {
            this.threadName = threadName;
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true){
                Goods goods = new Goods();
                try {
                    //模擬生產(chǎn)過程中的耗時(shí)操作
                    Thread.sleep(new Random().nextInt(100));
                    queue.put(goods);
                    System.out.println("【" + threadName + "】生產(chǎn)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    class Consumer extends Thread {
        private String threadName;
        private BlockingQueue<Goods> queue;

        public Consumer(String threadName, BlockingQueue<Goods> queue) {
            this.threadName = threadName;
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true){
                try {
                    Goods goods = queue.take();
                    System.out.println("【" + threadName + "】消費(fèi)了一個商品：【" + goods.toString() + "】，目前商品數(shù)量：" + queue.size());
                    //模擬消費(fèi)過程中的耗時(shí)操作
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Test
    public void test() {

        int maxSize = 5;
        BlockingQueue<Goods> queue = new LinkedBlockingQueue<>(maxSize);

        Thread producer1 = new ProducerConsumer3.Producer("生產(chǎn)者1", queue);
        Thread producer2 = new ProducerConsumer3.Producer("生產(chǎn)者2", queue);
        Thread producer3 = new ProducerConsumer3.Producer("生產(chǎn)者3", queue);

        Thread consumer1 = new ProducerConsumer3.Consumer("消費(fèi)者1", queue);
        Thread consumer2 = new ProducerConsumer3.Consumer("消費(fèi)者2", queue);


        producer1.start();
        producer2.start();
        producer3.start();
        consumer1.start();
        consumer2.start();

        while (true) {
        }
    }
}

要注意的地方：

如果使用LinkedBlockingQueue作為隊(duì)列實(shí)現(xiàn)，則可以實(shí)現(xiàn)：在同一時(shí)刻，既可以放入又可以取出，因?yàn)長inkedBlockingQueue內(nèi)部使用了兩個重入鎖，分別控制取出和放入。
如果使用ArrayBlockingQueue作為隊(duì)列實(shí)現(xiàn)，則在同一時(shí)刻只能放入或取出，因?yàn)锳rrayBlockingQueue內(nèi)部只使用了一個重入鎖來控制并發(fā)修改操作。

使用管道流實(shí)現(xiàn)

//TODO

無鎖的緩存框架: Disruptor

BlockingQueue 實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者模式簡單易懂，但是BlockingQueue并不是一個高性能的實(shí)現(xiàn)：它完全使用鎖和阻塞來實(shí)現(xiàn)線程之間的同步。在高并發(fā)的場合，它的性能并不是特別的優(yōu)越。（ConconcurrentLinkedQueue是一個高性能的隊(duì)列，但并不沒有實(shí)現(xiàn)BlockingQueue接口，即不支持阻塞操作）。

Disruptor是LMAX公司開發(fā)的高效的無鎖緩存隊(duì)列。它使用無鎖的方式實(shí)現(xiàn)了一個環(huán)形隊(duì)列，非常適合于實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者模式，如：事件和消息的發(fā)布。

//TODO 應(yīng)用場景的代碼實(shí)現(xiàn)

參考

Java 實(shí)現(xiàn)生產(chǎn)者 – 消費(fèi)者模型：各種實(shí)現(xiàn)方式的性能
高性能的生產(chǎn)者-消費(fèi)者：無鎖的實(shí)現(xiàn)：無鎖實(shí)現(xiàn)
Java生產(chǎn)者和消費(fèi)者模型的5種實(shí)現(xiàn)方式
生產(chǎn)者/消費(fèi)者問題的多種Java實(shí)現(xiàn)方式
Java阻塞隊(duì)列ArrayBlockingQueue和LinkedBlockingQueue實(shí)現(xiàn)原理分析：兩種常用阻塞隊(duì)列的區(qū)別

完整代碼示例在此