Java生产者消费者wait/notify实现详解

生产者消费者模式通过协调生产者和消费者对共享缓冲区的访问，实现多线程协作。1. 使用wait()/notifyAll()机制：当缓冲区满时生产者等待，空时消费者等待，通过notifyAll()唤醒线程避免死锁；2. 选择合适的阻塞队列：如ArrayBlockingQueue（有界队列适合稳定场景）、LinkedBlockingQueue（适合速度差异大场景）、PriorityBlockingQueue（优先级处理）、DelayQueue（延迟任务）和SynchronousQueue（传递性场景）；3. 其他实现方式：包括使用BlockingQueue简化代码、ReentrantLock与Condition提供更灵活控制。不同方法适用于不同需求，BlockingQueue适合简单实现，ReentrantLock适合复杂控制，而wait/notify是基础理解方式。

生产者消费者模式是一种经典的多线程协作模式，它巧妙地平衡了生产速度和消费速度，避免了资源浪费和数据丢失。在Java中，我们可以利用wait()和notify()/notifyAll()方法来实现这一模式。简单来说，生产者负责生产数据并放入共享缓冲区，消费者负责从缓冲区取出数据进行消费。

解决方案

实现生产者消费者模式的核心在于协调生产者和消费者对共享缓冲区的访问。以下是一个简单的Java实现：

import java.util.LinkedList;
import java.util.Queue;

public class ProducerConsumer {

    private static final int CAPACITY = 5;
    private final Queue buffer = new LinkedList<>();
    private final Object lock = new Object();

    class Producer implements Runnable {
        @Override
        public void run() {
            int i = 0;
            while (true) {
                synchronized (lock) {
                    try {
                        while (buffer.size() == CAPACITY) {
                            System.out.println("Buffer is full, Producer is waiting");
                            lock.wait(); // 缓冲区满，生产者等待
                        }

                        buffer.offer(i);
                        System.out.println("Produced: " + i);
                        i++;
                        lock.notifyAll(); // 通知消费者
                        Thread.sleep((long) (Math.random() * 100)); // 模拟生产时间
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    }

    class Consumer implements Runnable {
        @Override
        public void run() {
            while (true) {
                synchronized (lock) {
                    try {
                        while (buffer.isEmpty()) {
                            System.out.println("Buffer is empty, Consumer is waiting");
                            lock.wait(); // 缓冲区空，消费者等待
                        }

                        int value = buffer.poll();
                        System.out.println("Consumed: " + value);
                        lock.notifyAll(); // 通知生产者
                        Thread.sleep((long) (Math.random() * 200)); // 模拟消费时间
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }
    }

    public static void main(String[] args) {
        ProducerConsumer pc = new ProducerConsumer();
        new Thread(pc.new Producer()).start();
        new Thread(pc.new Consumer()).start();
        new Thread(pc.new Consumer()).start(); // 多个消费者
    }
}

这段代码中，buffer 是共享缓冲区，lock 是用于同步的锁对象。Producer 和 Consumer 分别是生产者和消费者线程。关键在于 wait() 和 notifyAll() 的使用：

• wait()：当缓冲区满（生产者）或空（消费者）时，线程调用 wait() 方法释放锁并进入等待状态。
• notifyAll()：当生产者生产了新的数据或消费者消费了数据后，调用 notifyAll() 方法唤醒所有等待的线程，让它们重新竞争锁。

为什么使用 notifyAll() 而不是 notify()？

虽然 notify() 也能唤醒一个等待的线程，但在生产者消费者模式中，使用 notifyAll() 更安全。 考虑以下情况：如果只有一个消费者线程在等待，notify() 可以唤醒它。但是，如果有多个消费者线程都在等待，而 notify() 唤醒的恰好也是一个消费者线程，那么这个线程可能会发现缓冲区仍然是空的，然后再次进入等待状态。这样就可能导致死锁。notifyAll() 确保所有等待的线程都有机会被唤醒并检查条件，从而避免死锁。

如何选择合适的阻塞队列？

Java提供了多种阻塞队列，例如 ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue、SynchronousQueue 等。选择合适的阻塞队列取决于具体的应用场景：

• ArrayBlockingQueue: 基于数组实现的有界阻塞队列，需要指定容量。适合于生产者和消费者速度相对稳定的场景。
• LinkedBlockingQueue: 基于链表实现的阻塞队列，可以选择有界或无界。无界队列可能导致内存溢出，需要谨慎使用。适合于生产者和消费者速度差异较大的场景。
• PriorityBlockingQueue: 具有优先级的无界阻塞队列。元素必须实现 Comparable 接口。适合于需要按照优先级处理任务的场景。
• DelayQueue: 延迟队列，元素需要在指定的延迟时间后才能被消费。元素必须实现 Delayed 接口。适合于实现定时任务的场景。
• SynchronousQueue: 同步队列，每个插入操作必须等待一个相应的移除操作，反之亦然。可以看作是一个容量为 0 的阻塞队列。适合于传递性场景，例如线程池。

选择阻塞队列时，需要考虑以下因素：

• 是否有界: 有界队列可以防止内存溢出，但可能导致生产者阻塞。无界队列则没有这个限制，但需要注意内存消耗。
• 是否需要优先级: 如果需要按照优先级处理任务，则需要选择 PriorityBlockingQueue。
• 是否需要延迟: 如果需要在指定的延迟时间后才能消费元素，则需要选择 DelayQueue。
• 并发性能: 不同阻塞队列的并发性能可能有所不同，需要根据实际情况进行选择。

除了wait/notify，还有哪些实现生产者消费者模式的方式？

除了 wait()/notify() 机制，Java还提供了其他方式来实现生产者消费者模式，例如：

• BlockingQueue: Java并发包 java.util.concurrent 提供的阻塞队列接口。它简化了生产者消费者模式的实现，无需手动管理锁和等待/通知机制。例如上面的代码可以改为使用 ArrayBlockingQueue。

  import java.util.concurrent.ArrayBlockingQueue;
  import java.util.concurrent.BlockingQueue;
  
  public class ProducerConsumerBlockingQueue {
  
      private static final int CAPACITY = 5;
      private final BlockingQueue buffer = new ArrayBlockingQueue<>(CAPACITY);
  
      class Producer implements Runnable {
          @Override
          public void run() {
              int i = 0;
              while (true) {
                  try {
                      buffer.put(i); // 阻塞直到队列有空间
                      System.out.println("Produced: " + i);
                      i++;
                      Thread.sleep((long) (Math.random() * 100));
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  }
              }
          }
      }
  
      class Consumer implements Runnable {
          @Override
          public void run() {
              while (true) {
                  try {
                      int value = buffer.take(); // 阻塞直到队列有元素
                      System.out.println("Consumed: " + value);
                      Thread.sleep((long) (Math.random() * 200));
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  }
              }
          }
      }
  
      public static void main(String[] args) {
          ProducerConsumerBlockingQueue pc = new ProducerConsumerBlockingQueue();
          new Thread(pc.new Producer()).start();
          new Thread(pc.new Consumer()).start();
          new Thread(pc.new Consumer()).start();
      }
  }

  使用 BlockingQueue 可以大大简化代码，并提高可读性。put() 方法在队列满时会阻塞，take() 方法在队列空时会阻塞，无需手动处理等待和通知。

• ReentrantLock 和 Condition: ReentrantLock 提供了比 synchronized 更灵活的锁机制，Condition 则提供了比 wait()/notify() 更精细的线程等待和通知机制。 可以实现更复杂的同步逻辑。

  import java.util.LinkedList;
  import java.util.Queue;
  import java.util.concurrent.locks.Condition;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  public class ProducerConsumerReentrantLock {
  
      private static final int CAPACITY = 5;
      private final Queue buffer = new LinkedList<>();
      private final Lock lock = new ReentrantLock();
      private final Condition notFull = lock.newCondition();
      private final Condition notEmpty = lock.newCondition();
  
      class Producer implements Runnable {
          @Override
          public void run() {
              int i = 0;
              while (true) {
                  lock.lock();
                  try {
                      while (buffer.size() == CAPACITY) {
                          System.out.println("Buffer is full, Producer is waiting");
                          notFull.await(); // 缓冲区满，生产者等待
                      }
  
                      buffer.offer(i);
                      System.out.println("Produced: " + i);
                      i++;
                      notEmpty.signalAll(); // 通知消费者
                      Thread.sleep((long) (Math.random() * 100));
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  } finally {
                      lock.unlock();
                  }
              }
          }
      }
  
      class Consumer implements Runnable {
          @Override
          public void run() {
              while (true) {
                  lock.lock();
                  try {
                      while (buffer.isEmpty()) {
                          System.out.println("Buffer is empty, Consumer is waiting");
                          notEmpty.await(); // 缓冲区空，消费者等待
                      }
  
                      int value = buffer.poll();
                      System.out.println("Consumed: " + value);
                      notFull.signalAll(); // 通知生产者
                      Thread.sleep((long) (Math.random() * 200));
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  } finally {
                      lock.unlock();
                  }
              }
          }
      }
  
      public static void main(String[] args) {
          ProducerConsumerReentrantLock pc = new ProducerConsumerReentrantLock();
          new Thread(pc.new Producer()).start();
          new Thread(pc.new Consumer()).start();
          new Thread(pc.new Consumer()).start();
      }
  }

  在这个例子中，notFull 和 notEmpty 是两个 Condition 对象，分别用于生产者和消费者的等待和通知。await() 方法类似于 wait()，signalAll() 方法类似于 notifyAll()。 使用 ReentrantLock 和 Condition 可以实现更细粒度的控制，例如可以只唤醒特定的生产者或消费者。

选择哪种方式取决于具体的需求。 如果只是简单的生产者消费者模式，使用 BlockingQueue 是最简单和推荐的方式。 如果需要更灵活的控制，可以使用 ReentrantLock 和 Condition。 而 wait()/notify() 机制则是最基础的方式，理解它可以帮助我们更好地理解并发编程的原理。