java - Java 中通知和等待的顺序

Question

我正在尝试编写生产者消费者代码。下面是我写的原始代码。

Stack<Integer> buffer = new Stack<>();
volatile int i = 1;

class Consumer implements Runnable {

    @Override
    public void run() {
        while(true){
            synchronized (buffer) {
                System.out.println("Consumer taking lock : " + Thread.currentThread().getName());
                while(buffer.isEmpty()){
                    try{
                        System.out.println("Consumer releasing lock :" + Thread.currentThread().getName());
                        buffer.wait();
                        System.out.println("Consumer woken up :" + Thread.currentThread().getName());
                    } catch(InterruptedException ie){
                        ie.printStackTrace();
                    }
                }
                System.out.println(buffer.pop());
                buffer.notify();
            }
        }
    }
}

class Producer implements Runnable {

    @Override
    public void run() {
        while(true){
            synchronized (buffer) {
                System.out.println("Producer taking lock : " + Thread.currentThread().getName());
                while(!buffer.isEmpty()){
                    try {
                        System.out.println("Producer going into wait set :" + Thread.currentThread().getName());
                        buffer.wait();
                        System.out.println("Producer woken up :" + Thread.currentThread().getName());
                    } catch (InterruptedException ie) {
                        ie.printStackTrace();
                    }
                }
                buffer.push(i);
                i++;
                buffer.notify();
            }
        }
    }

}

public static void main(String[] args) {

    ProducerConsumerUnitBuffer obj = new ProducerConsumerUnitBuffer();
    Thread producerThread1 = new Thread(obj.new Consumer());
    Thread consumerThread1 = new Thread(obj.new Producer());
    Thread producerThread2 = new Thread(obj.new Consumer());
    Thread consumerThread2 = new Thread(obj.new Producer());
    Thread producerThread3 = new Thread(obj.new Consumer());
    Thread consumerThread3 = new Thread(obj.new Producer());
    Thread producerThread4 = new Thread(obj.new Consumer());
    Thread consumerThread4 = new Thread(obj.new Producer());
    Thread producerThread5 = new Thread(obj.new Consumer());
    Thread consumerThread5 = new Thread(obj.new Producer());
    Thread producerThread6 = new Thread(obj.new Consumer());
    Thread consumerThread6 = new Thread(obj.new Producer());
    Thread producerThread7 = new Thread(obj.new Consumer());
    Thread consumerThread7 = new Thread(obj.new Producer());
    Thread producerThread8 = new Thread(obj.new Consumer());
    Thread consumerThread8 = new Thread(obj.new Producer());
    Thread producerThread9 = new Thread(obj.new Consumer());
    Thread consumerThread9 = new Thread(obj.new Producer());
    Thread producerThread10 = new Thread(obj.new Consumer());
    Thread consumerThread10 = new Thread(obj.new Producer());

    producerThread1.start();
    consumerThread1.start();
    producerThread2.start();
    consumerThread2.start();
    producerThread3.start();
    consumerThread3.start();
    producerThread4.start();
    consumerThread4.start();
    producerThread5.start();
    consumerThread5.start();
    producerThread6.start();
    consumerThread6.start();
    producerThread7.start();
    consumerThread7.start();
    producerThread8.start();
    consumerThread8.start();
    producerThread9.start();
    consumerThread9.start();
    producerThread10.start();
    consumerThread10.start();

}

此代码总是停止。尽管应用程序不会终止，但它会停止打印任何内容，这意味着没有线程进入同步块(synchronized block)。

尽管如此，当我使用notifyAll()而不是notify()时，代码工作得很好。

编辑

根据建议，我尝试更改代码，以便生产者和消费者有 2 个单独的对象可以锁定。此外，当对象放入缓冲区时，生产者会通知消费者。

public class ProducerConsumerDifferentObjects {

Stack<Integer> buffer = new Stack<>();
Boolean producerLockingObject = Boolean.FALSE;
Boolean consumerLockingObject = Boolean.TRUE;
volatile int i = 1;

class Consumer implements Runnable {
    @Override
    public void run() {
        while (true) {
            synchronized (consumerLockingObject) {
                while (buffer.isEmpty()) {
                    try {
                        consumerLockingObject.wait();
                    } catch (InterruptedException ie) {
                        ie.printStackTrace();
                    }
                }
                System.out.println(buffer.pop());
                consumerLockingObject.notify();
            }
        }
    }
}

class Producer implements Runnable {
    @Override
    public void run() {
        while (true) {
            synchronized (producerLockingObject) {
                while (!buffer.isEmpty()) {
                    try {
                        producerLockingObject.wait();
                    } catch (InterruptedException ie) {
                        ie.printStackTrace();
                    }
                }
                buffer.push(i);
                i++;
                producerLockingObject.notify();
                synchronized (consumerLockingObject) {
                    consumerLockingObject.notify();
                }
            }
        }
    }
}

public static void main(String[] args) {

    ProducerConsumerDifferentObjects obj = new ProducerConsumerDifferentObjects();
    Thread producerThread1 = new Thread(obj.new Consumer());
    Thread consumerThread1 = new Thread(obj.new Producer());
    Thread producerThread2 = new Thread(obj.new Consumer());
    Thread consumerThread2 = new Thread(obj.new Producer());
    Thread producerThread3 = new Thread(obj.new Consumer());
    Thread consumerThread3 = new Thread(obj.new Producer());
    Thread producerThread4 = new Thread(obj.new Consumer());
    Thread consumerThread4 = new Thread(obj.new Producer());
    Thread producerThread5 = new Thread(obj.new Consumer());
    Thread consumerThread5 = new Thread(obj.new Producer());
    Thread producerThread6 = new Thread(obj.new Consumer());
    Thread consumerThread6 = new Thread(obj.new Producer());
    Thread producerThread7 = new Thread(obj.new Consumer());
    Thread consumerThread7 = new Thread(obj.new Producer());
    Thread producerThread8 = new Thread(obj.new Consumer());
    Thread consumerThread8 = new Thread(obj.new Producer());
    Thread producerThread9 = new Thread(obj.new Consumer());
    Thread consumerThread9 = new Thread(obj.new Producer());
    Thread producerThread10 = new Thread(obj.new Consumer());
    Thread consumerThread10 = new Thread(obj.new Producer());

    producerThread1.start();
    consumerThread1.start();
    producerThread2.start();
    consumerThread2.start();
    producerThread3.start();
    consumerThread3.start();
    producerThread4.start();
    consumerThread4.start();
    producerThread5.start();
    consumerThread5.start();
    producerThread6.start();
    consumerThread6.start();
    producerThread7.start();
    consumerThread7.start();
    producerThread8.start();
    consumerThread8.start();
    producerThread9.start();
    consumerThread9.start();
    producerThread10.start();
    consumerThread10.start();
}

}

Answer 1

在多个生产者/多个消费者(MPMC)场景中，您尝试使用单个对象(缓冲区) 用于通知双方消费者和生产者。这就是您最终陷入停滞的原因:一个消费者通知另一个消费者，而不是通知生产者。 (或者生产者通知另一个生产者)。

考虑以下可能的事件顺序:

1. 使用者 1 等待空缓冲区。
2. 使用者 2 等待空缓冲区。
3. 生产者生产一个元素，并通知消费者 1。然后生产者等待非空缓冲区。
4. 消费者 1 醒来，消耗缓冲区中的唯一元素，通知消费者 2，并等待空缓冲区。
5. 消费者 2 醒来，但立即等待，因为缓冲区为空。

因此所有消费者和生产者都处于等待状态。

拥有更多的消费者和/或生产者并没有帮助:它们仍然有可能处于等待状态。

<小时/>

解决问题的可能方法:

1. 拥有不同的对象来通知消费者和生产者。在Java中，您可以创建两个Condition来自单个锁的对象，通知消费者的一个条件，另一个 - 通知生产者的条件。

2. 另外，生产者与自身同步，消费者与自身同步。因此至多单个生产者和单个消费者可能会等待公共(public)对象。

这两种方法都消除了消费者-消费者或生产者-生产者通知的可能性。

<小时/>

在缓冲区容量为 2 个或更多元素的情况下，将生产者的等待条件从“缓冲区非空”更改为“缓冲区已满”可能会有所帮助。消费者-消费者通知仍然是可能的，但可能会避免完全停止。但这会非常棘手。