你真的了解Wait 和Notify吗？

前言

   大家对生产者和消费者模式很熟悉吧，这个模式很好理解，也在工程实践中经常被使用到。学习Java的同学有很大一部分使用Java语言实现过生产者和消费者模式，我就是其中之一。大概思路：有个盛装数据的容器(list)即缓冲区,一个往容器里放数据即生产者,一个从容器中取数据即消费者。但当容器满的时候，生产者就不能往里放东西了，此时需要等待缓冲区不满，即有消费者从容器中取出数据了，这就需要一个等待和通知的功能了。这个功能我当时是直接使用Java 中 Object 定义的wait、notify和notifyAll来实现的。现在回忆起来也只能记得这么多。今天的这篇文章会对通知与等待有个比较深入的探索，以便写出更安全的代码。

有问题却一眼看不出问题的消费者和生产者模式

   下文将使用Java 实现一个简单的消费者和生产者模式，代码如下：

public class Producer {
    //缓冲区 
    private CircleQueue<String> cache;


    public Producer(CircleQueue<String> cache) {
        this.cache = cache;
    }
    
    public synchronized void produce(String e) throws InterruptedException {
        if (cache.isFull()) {
            cache.wait();
        }
        cache.put(e);
        cache.notifyAll();
    }
}

public class Consumer {
    private CircleQueue<String> cache;
    public Consumer(CircleQueue<String> container) {
        this.cache = container;
    }
    public synchronized void consume() throws InterruptedException {
        if (cache .isEmpty()) {
            lock.wait();
        }
        System.out.println(
                String.format("thread:%s,consume a element:%s",
                        Thread.currentThread().getName(),
                        cache .take()));
        cache.notifyAll();
    }
}

上面的代码很短，我就没有注释。大家看了一遍，有没有看出来上面代码运行会报错？运行一下测试代码(见最下面)大家就明白了：
结果：

调用notifyAll 但是并没有拥有指定的监视器
生产者执行produce 函数的时候，执行到cache.notifyAll() 会抛出这个异常，这个异常代表什么意思呢？
我们来看看IllegalMonitorStateException 的注释：
Thrown to indicate that a thread has attempted to wait on an object's monitor or to notify other threads waiting on an object's monitor without owning the specified monitor.
上面这个注释, 它想表达两个意思
1> 在不拥有当前对象监视器的情况下调用当前对象的wait方法
2> 在不拥有当前对象监视器的情况下调用当前对象的notify和notifyAll
上面报错的原因就是因为满足了第2条,其实同样 cache.wait() 这里也有问题。下面我们来改下代码：

public void produce(String e) throws InterruptedException {
        synchronized (cache) {
            if (cache.isFull()) {
                cache .wait();
            }
            cache.put(e);
            cache.notifyAll();
        }
    }

public  void consume() throws InterruptedException {
        synchronized (cache) {
            if (container.isEmpty()) {
                cache.wait();
            }
            System.out.println(
                    String.format("thread:%s,consume a element:%s",
                            Thread.currentThread().getName(),
                            cache.take()));
            cache.notifyAll();
        }
    }

上面只粘贴了主要的代码，synchronized 关键字修饰的是cache,因此当线程进来的时候获取的是cache对象的监视器，因此到下面无论执行cache.wait()还是cache.notifyAll() 都不会抛出异常。当然上面的代码还是有问题的，不知道大家有没有看出来？我们来执行下测试代码，发现报错了，但是也有可能你执行的时候并没有报错，因为这是多线程，存在很多偶然性，多运行几遍你就会发现你中奖了，报错如下：

wait时释放锁，被唤醒后重新获取锁

上面报错的是Consumer 中的cache.take()处抛的异常，cache对象的类型CircleQueue 是我自己实现的一个环形队列,抛出该异常是因为队列已空，如果这样大家就奇怪了，能执行到cache.take() 不是因为队列不为空才唤醒当前线程的吗？
  我先介绍下当某个线程调用wait的时候发生了什么？
它会释放调用对象上的监视器即锁，然后进入一个条件等待队列中等待被唤醒，此时有别的线程改变了状态(eg: 队列为空，队列已满),然后调用notifyAll(),条件等待队列中的线程再次获取调用对象上的监视器，然后继续向下执行。画一个草图给大家理解下：

锁与条件变量

   在一个线程被唤醒到获取锁的这个时间里，可能有另外一个线程改变了状态 ( 往队列插入元素或者从队列中取出元素 ), 然后该线程往下执行的时候发现缓冲区为空或者已满这种现象。大家估计也想到了解决办法，就是被唤醒再次获取锁之后再判断一次状态，加一个循环就搞定了，代码如下：

 public void consume() throws InterruptedException {
        synchronized (cache) {
          // 这里把if改为while 就可以了
            while (cache.isEmpty()) {
                cache.wait();
            }
            System.out.println(
                    String.format("thread:%s,consume a element:%s",
                            Thread.currentThread().getName(),
                            cache.take()));
            cache.notifyAll();
        }
    }

这里限于篇幅，我只粘贴了消费者的代码,生产者同样的地方改一下就可以。多运行几次测试代码，都正常运行，并无线程安全的问题。

后记

   有可能大家一开始学习的版本就是我最终实现的版本，可是大家知道是怎么来的吗？知其所以然很重要，你不需要去硬记很多东西。谢谢大家观看，麻烦点个赞和关注一下。

附录

测试代码

 public static void main(String[] args) {
        CircleQueue<String> cache = new CircleQueue<>(10);
        Object lock = new Object();
        for (int j = 0; j < 2; j++) {
            new Thread(() -> {
                Producer producer = new Producer(cache);
                for (int i = 0; i < 10; i++) {
                    try {
                        producer.produce("数据" + i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }

        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                Consumer consumer = new Consumer(cache);
                while (true) {
                    try {
                        consumer.consume();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            }).start();
        }
    }