关于使用阻塞队列方法的Java生产者和消费者模式

Question

我正在研究有关Java中线程的生产者和消费者设计模式，最近我在Java 5中进行了探索，引入Java
5中引入了BlockingQueue数据结构。现在，它变得更加简单，因为BlockingQueue通过引入阻塞方法隐式地提供了此控件。
put（）和take（）。现在，您无需使用等待和通知即可在生产者和消费者之间进行通信。如果有界队列，则如果Queue已满，BlockingQueue
put（）方法将阻塞；如果Queue为空，则take（）方法将阻塞。在下一节中，我们将看到Producer
Consumer设计模式的代码示例。我已经开发了以下程序，但也请让我知道waut（）和notify（）的旧方法，我也想使用旧方法开发相同的逻辑

民间请告知如何实现，经典方式是使用wait（）和notify（）方法在Producer和Consumer线程之间进行通信，并在诸如完整队列和空队列之类的个别条件下阻止它们中的每一个…？

    import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ProducerConsumerPattern {

    public static void main(String args[]){

     //Creating shared object
     BlockingQueue sharedQueue = new LinkedBlockingQueue();

     //Creating Producer and Consumer Thread
     Thread prodThread = new Thread(new Producer(sharedQueue));
     Thread consThread = new Thread(new Consumer(sharedQueue));

     //Starting producer and Consumer thread
     prodThread.start();
     consThread.start();
    }

}

//Producer Class in java
class Producer implements Runnable {

    private final BlockingQueue sharedQueue;

    public Producer(BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        for(int i=0; i<10; i++){
            try {
                System.out.println("Produced: " + i);
                sharedQueue.put(i);
            } catch (InterruptedException ex) {
                Logger.getLogger(Producer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }

}

//Consumer Class in Java
class Consumer implements Runnable{

    private final BlockingQueue sharedQueue;

    public Consumer (BlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        while(true){
            try {
                System.out.println("Consumed: "+ sharedQueue.take());
            } catch (InterruptedException ex) {
                Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }


}

Output:
Produced: 0
Produced: 1
Consumed: 0
Produced: 2
Consumed: 1
Produced: 3
Consumed: 2
Produced: 4
Consumed: 3
Produced: 5
Consumed: 4
Produced: 6
Consumed: 5
Produced: 7
Consumed: 6
Produced: 8
Consumed: 7
Produced: 9
Consumed: 8
Consumed: 9

Answer 1

我可以在睡觉时（或至少我认为可以）进行等待通知。Java
1.4源提供了所有这些的漂亮示例，但是他们已经切换到使用原子来完成所有事情，并且现在变得更加复杂。尽管其他方法可以使您免受并发危险的困扰，并且可以简化代码，但wait-
notify确实提供了灵活性和强大的功能。

为此，您需要一些字段，例如：

private final ConcurrentLinkedQueue<Intger>  sharedQueue =
                                                    new ConcurrentLinkedQueue<>();
private volatile   boolean  waitFlag = true;

您的Producer.run看起来像这样：

public void run()  {
    for (int i = 0; i < 100000, i++)  {
        System.out.println( "Produced: " + i );
        sharedQueue.add( new Integer( i ) );
        if (waitFlag)       // volatile access is cheaper than synch.
            synchronized (sharedQueue)  { sharedQueue.notifyAll(); }
    }
}

和Consumer.run：

public void run()  {
    waitFlag = false;
    for (;;)  {
        Integer  ic = sharedQueue.poll();
        if (ic == null)  {
            synchronized (sharedQueue)  {
                waitFlag = true;
                // An add might have come through before waitFlag was set.
                ic = sharedQueue.poll();
                if (ic == null)  {
                    try  { sharedQueue.wait(); }
                    catch (InterruptedException ex)  {}
                    waitFlag = false;
                    continue;
                }
                waitFlag = true;
            }
        }
        System.out.println( "Consumed: " + ic );
    }
}

这样可以使同步保持在最低水平。如果一切顺利，那么每次添加时只有一个不稳定的字段可以看一看。您应该能够同时运行任意数量的生产者。（消费者会比较棘手-
waitFlag您必须放弃。）您可以将另一个对象用于wait / notifyAll。