生产者-消费者（wait-notifyAll）

black-sedan-parked-under-tree-1841120.jpg

简介：

生产者、消费者真的是老生常谈了，很多地方都会用到这种思想，可见其重要程度。所以，还是有必要了解一下它的最原始的实现方式的，也就是利用Object的wait和notifyAll方法。

源码：

/**
     * Causes the current thread to wait until another thread invokes the
     * {@link java.lang.Object#notify()} method or the
     * {@link java.lang.Object#notifyAll()} method for this object.
     * In other words, this method behaves exactly as if it simply
     * performs the call {@code wait(0)}.
     * <p>
     * The current thread must own this object's monitor. The thread
     * releases ownership of this monitor and waits until another thread
     * notifies threads waiting on this object's monitor to wake up
     * either through a call to the {@code notify} method or the
     * {@code notifyAll} method. The thread then waits until it can
     * re-obtain ownership of the monitor and resumes execution.
     * <p>
     * As in the one argument version, interrupts and spurious wakeups are
     * possible, and this method should always be used in a loop:
     * <pre>
     *     synchronized (obj) {
     *         while (&lt;condition does not hold&gt;)
     *             obj.wait();
     *         ... // Perform action appropriate to condition
     *     }
     * </pre>
     * This method should only be called by a thread that is the owner
     * of this object's monitor. See the {@code notify} method for a
     * description of the ways in which a thread can become the owner of
     * a monitor.
     *
     * @throws  IllegalMonitorStateException  if the current thread is not
     *               the owner of the object's monitor.
     * @throws  InterruptedException if any thread interrupted the
     *             current thread before or while the current thread
     *             was waiting for a notification.  The <i>interrupted
     *             status</i> of the current thread is cleared when
     *             this exception is thrown.
     * @see        java.lang.Object#notify()
     * @see        java.lang.Object#notifyAll()
     */
    public final void wait() throws InterruptedException {
        wait(0);
    }

    /**
     * Wakes up all threads that are waiting on this object's monitor. A
     * thread waits on an object's monitor by calling one of the
     * {@code wait} methods.
     * <p>
     * The awakened threads will not be able to proceed until the current
     * thread relinquishes the lock on this object. The awakened threads
     * will compete in the usual manner with any other threads that might
     * be actively competing to synchronize on this object; for example,
     * the awakened threads enjoy no reliable privilege or disadvantage in
     * being the next thread to lock this object.
     * <p>
     * This method should only be called by a thread that is the owner
     * of this object's monitor. See the {@code notify} method for a
     * description of the ways in which a thread can become the owner of
     * a monitor.
     *
     * @throws  IllegalMonitorStateException  if the current thread is not
     *               the owner of this object's monitor.
     * @see        java.lang.Object#notify()
     * @see        java.lang.Object#wait()
     */
    public final native void notifyAll();

思路：

首先要想的是生产者和消费者分别需要用java里面的什么来实现，毫无疑问，肯定要用线程了，因为它是最基本的调度单元，可以在其run方法里面处理自定义逻辑。确定了线程之后，还要找一个地方来存储生产者生产出来的产品，首先这个容器要有序的，那就用最简单的ArrayList了。有的时候为了测试方便，我们有可能要协调多个线程，可能需要借助于CountDownLatch来实现。

举例1：

这个例子展示的是，开启多个生产者生产一定量的产品放进队列，等所有的生产者都生产完之后，打印生产了多少个产品。然后，多个消费者再开始消费，消费完之后，再打印，容器里面还剩余多少个产品。以此体验生产和消费的思想,大家可以在自己的IDE上运行一下。

    @Test
    public void test01() throws InterruptedException {
        CountDownLatch countDownLatchProducer = new CountDownLatch(15);
        CountDownLatch countDownLatchConsumer = new CountDownLatch(6);
        for (int i = 0; i < 3; i++) {//定义3个线程
            new Thread(new Producer(i,countDownLatchProducer)).start();
        }
//        Thread.sleep(20000);
        countDownLatchProducer.await();//当所有的生产者子线程的任务都完成之后，就会恢复继续向下执行。
        System.out.println("消费前大小："+Queue.getList().size());
        for (int i = 0; i < 3; i++) {
            new Thread(new Consumer(countDownLatchConsumer)).start();
        }
//        Thread.sleep(20000);
        countDownLatchConsumer.await();//当所有的消费者子线程的任务都完成之后，就会恢复继续向下执行。
        System.out.println("消费后大小："+Queue.getList().size());
    }

    class Producer implements Runnable {

        private int i;
        private CountDownLatch countDownLatch;

        public Producer(int i,CountDownLatch countDownLatch) {
            this.i = i;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 5; i++) {//每个生产者线程每次生产5个对象
                    Queue.put(new Bread(i));
                    countDownLatch.countDown();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    static class Consumer implements Runnable {

        private CountDownLatch countDownLatch;

        public Consumer(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 2; i++) {//每个消费者线程一次消费2个对象
                    Bread bread = Queue.get();
                    countDownLatch.countDown();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

public class Queue {

    private final static List<Bread> list = new ArrayList<>(15);

    public static List getList() {
        return list;
    }

    static void put(Bread bread) throws InterruptedException {//进队列
        synchronized (list) {
            while (list.size() >= 15) {//限制容器大小最大值为15
                list.wait();
            }
            list.add(bread);
            list.notifyAll();
        }
    }

    static Bread get() throws InterruptedException {//出队列
        Bread bread = null;
        synchronized (list) {
            if (list.size() == 0)
                list.wait();
            bread = list.remove(list.size() - 1);
            list.notifyAll();
        }
        return bread;
    }
}

package com.hao.hellolearn.jdk.producerconsumer;

public class Bread {

    private int value;

    public Bread(int value) {
        this.value = value;
    }

    public int getValue() {
        return value;
    }

    public void setValue(int value) {
        this.value = value;
    }
}

举例2：

这个例子展示的是：当容器的容量不足以装下生产者生产的任务的时候，生产者的线程就会阻塞在put方法上，直到消费者开始消费容器中的任务，生产者线程才会恢复执行，可以通过sleep方法来协助测试。

public class WaitNotifyImpl {

    @Test
    public void test01() throws InterruptedException {
        for (int i = 0; i < 3; i++) {//定义3个线程
            new Thread(new Producer(i)).start();
        }
        Thread.sleep(20000);
        System.out.println("消费前大小："+Queue.getList().size());
        for (int i = 0; i < 3; i++) {
            new Thread(new Consumer()).start();
        }
        Thread.sleep(20000);
        System.out.println("消费后大小："+Queue.getList().size());
    }

    class Producer implements Runnable {

        private int i;

        public Producer(int i) {
            this.i = i;
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 5; i++) {//每个生产者线程每次生产5个对象
                    Queue.put(new Bread(i));
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    static class Consumer implements Runnable {

        public Consumer() {
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 2; i++) {//每个消费者线程一次消费2个对象
                    Bread bread = Queue.get();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

当第一个sleep执行的时候，可以在cmd里面分别执行jps -l 查看运行中的进程的pid，然后用jstack -l pid 来打印出这个进程的堆栈信息，可以看到线程阻塞在object monitor上面。1处的是生产者线程的状态是WAITING是有object.wait 引起的，2处的是TIMED_WAITING是由sleep引起的。

生产者-消费者-jstack.png

总结：

生产者-消费者 生产出来的产品放在一个容器里面，消费者从容器里面取出来；如果生产速率慢，消费速度快，那么当容器为空的时候，消费者会阻塞在当前容器（什么事也做不了，静静的等待），直到容器里面被放进一个产品，然后争抢着。如果消费速率慢，生产速度快，那么当容器满了的时候生产者会阻塞在当前容器（什么事也做不了，静静的等待），直到容器里面的某一个产品被消费掉。它的本质其实就是线程之间的协作。
觉察即自由。