具有多个消费者的生产者使用 notify() 失败

Posted 2023-02-16

【中文标题】具有多个消费者的生产者使用 notify() 失败

【英文标题】：Producer with multiple consumer failing with notify()

【发布时间】：2018-05-04 18:50:06

【问题描述】：

我有这段代码，它是在共享 Q 类对象上使用单个生产者和多个消费者模拟生产者消费者问题。我在这里使用 notify() 而不是 notifyAll()，因为我需要了解为什么此代码会进入死锁或无限等待状态。

我的观点是：如果有一个生产者和多个消费者，那么 notify() 将只调用一个处于等待状态的线程，而 rest 将保持在 wait() 状态。然后生产者将再次恢复，因此代码将继续执行。

观察：这里所有线程的生产者和消费者都处于无限等待状态。代码如下：

public class ProdConsProb 
    public static void main(String[] args) 
        Q q = new Q();
        Thread producerThread = new Thread(new Producer(q), "producerThread");
        Thread consumerThread = new Thread(new Consumer(q), "Consumer1");
        Thread consumerAnotherThread = new Thread(new Consumer(q), "Consumer2");
        Thread consumerYetAnotherThread = new Thread(new Consumer(q), "Consumer3");
        producerThread.start();
        consumerThread.start();
        consumerAnotherThread.start();
        consumerYetAnotherThread.start();
    


class Producer implements Runnable 
    Q q;

    public Producer(Q q) 
        this.q = q;
    

    @Override
    public void run() 
        int i = 0;
        while (true)
            try 
                q.setN(i++);
             catch (InterruptedException e) 
                e.printStackTrace();
            

    



class Consumer implements Runnable 
    Q q;

    public Consumer(Q q) 
        this.q = q;
    

    @Override
    public void run() 
        while (true)
            try 
                q.getN();
             catch (InterruptedException e) 
                // TODO Auto-generated catch block
                e.printStackTrace();
            
    



class Q 
    private int n = 0;
    boolean valueSet = false;

    public synchronized int getN() throws InterruptedException 
        while (!valueSet) 
        
            wait();
        
        valueSet = false;
        notify();
        return n;
    

    public synchronized void setN(int n) throws InterruptedException 
        while (valueSet == true)
        
            wait();
        
        this.n = n;
        valueSet = true;
        notify();

    

我添加了一些 sysouts 。生成的日志如下所示：

producerThread :: SetN : Valueset is false
producerThread :: Producer inserted 0
producerThread :: SetN : Valueset after is true
producerThread :: SetN : Valueset is true
producerThread wait() ------------ Active 
producerThread :: SetN :Wait() Valueset is true
Consumer3  Start :: GetN : Valueset is true
Consumer3 :: Consumer read 0
Consumer3  End :: GetN : Valueset after is false
Consumer3  Start :: GetN : Valueset is false
Consumer3 wait() ------------ Active 
Consumer3 :: GetN :Wait() Valueset is false
Consumer2  Start :: GetN : Valueset is false
Consumer2 wait() ------------ Active 
Consumer2 :: GetN :Wait() Valueset is false
Consumer1  Start :: GetN : Valueset is false
Consumer1 wait() ------------ Active 
Consumer1 :: GetN :Wait() Valueset is false
producerThread wait()   ------------- left 
producerThread :: Producer inserted 1
producerThread :: SetN : Valueset after is true
producerThread :: SetN : Valueset is true
producerThread wait() ------------ Active 
 -->>   producerThread :: SetN :Wait() Valueset is true
        Consumer3 wait() left 
        Consumer3 :: Consumer read 1
        Consumer3  End :: GetN : Valueset after is false
        Consumer3  Start :: GetN : Valueset is false
        Consumer3 wait() ------------ Active 
        Consumer3 :: GetN :Wait() Valueset is false
 ????   Consumer2 wait() left 
        Consumer2 wait() ------------ Active 
        Consumer2 :: GetN :Wait() Valueset is false

这里奇怪的是一旦生产者在插入1后通知，消费者3读取数据并通知生产者。现在生产者 3 必须从它的 wait() 触发，但 customer2 线程离开它的 wait() 并返回到 wait() 。

注意：此代码可与 notifyAll() 一起使用，但我正在寻找它与 notify() 失败的原因。

【参考方案1】：

它失败是因为生产者和消费者在同一个监视器上等待，而内在锁不支持单独的条件。如果通知发生，生产者或消费者都可以得到它。但是给定的通知将仅适用于其中一个。当一个人收到一个只有另一个人可以采取行动的通知时，该通知就没有任何好处：被通知的线程醒来，发现它正在等待的条件仍然是错误的，然后回到等待状态。

如果你看 ArrayBlockingQueue，它是用 ReentrantLock 实现的，有单独的条件，一个给消费者，一个给生产者，这样就不会发生这种错误。

【讨论】：

这确实有道理。我读到了 ReentrantLock 和他们的实现方式暗示了同样的事情。谢谢你。我会尝试使用 ReentrantLock 来实现上面的，然后想出一个解决方案