用Java原生的 wait() notify() notifyAll() 来实现

• Java的wait() notify() notifyAll() 是用来实现协同的，因此要与用来实现同步synchronized联合使用。

wait() 源码详解

• 首先一个线程想要调用wait() 方法，这个线程需要拥有对象的monitor，也就是synchronized锁；
• 在调用wait()方法后，这个线程将加入对象的等待队列并且释放锁；
• 等待另一个持有锁的线程通过notify() 或者notifyAll() 唤醒该线程;
• 当这个线程重新获取锁以后继续执行。
• 虚假唤醒：wait()要写进一个循环里，为了防止虚假唤醒的情况。

/**
     * 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.

notify() 源码详解

• 任意选择一个等待的线程唤醒；
• 被唤醒的线程只有在重新竞争到锁以后才可以继续执行；
• 调用notify() 的线程必须拥有当前对象的锁，否则会抛出IllegalMonitorStateException异常。

 /**
     * Wakes up a single thread that is waiting on this object's
     * monitor. If any threads are waiting on this object, one of them
     * is chosen to be awakened. The choice is arbitrary and occurs at
     * the discretion of the implementation. A thread waits on an object's
     * monitor by calling one of the {@code wait} methods.
     * <p>
     * The awakened thread will not be able to proceed until the current
     * thread relinquishes the lock on this object. The awakened thread will
     * compete in the usual manner with any other threads that might be
     * actively competing to synchronize on this object; for example, the
     * awakened thread enjoys 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.

package thread;

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import java.util.function.Consumer;


//wait notify
public class ProducerConsumer1 {
    public static void main(String[] args) throws InterruptedException {
        Queue<Integer> queue = new LinkedList<>();
        Producer t1 = new Producer(queue);
        Consumer t2 = new Consumer(queue);
        t1.start();
        t2.start();
        t1.join();
        t2.join();
    }

    static class Producer extends Thread {
        Queue<Integer> queue;

        Producer(Queue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            synchronized (queue) {
                while (!queue.isEmpty()) {
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                int tmp = new Random().nextInt();
                queue.offer(tmp);
                System.out.println("Producing " + tmp);
                queue.notifyAll();
            }
        }

    }

    //wait源码
    //synchronized （class）的时候报错，因为调用wait（）方法的线程必须持有当前对象的锁，如果用class对象当锁，该线程就没有持有当前对象的锁。
    static class Consumer extends Thread {
        Queue<Integer> queue;

        Consumer(Queue<Integer> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            synchronized (queue) {
                while (queue.isEmpty()) {
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
                int tmp = queue.poll();
                System.out.println("Consuming " + tmp);
                queue.notifyAll();
            }
        }

    }
}