java - 生产者和消费者在 Java 中使用锁

Question

我从 here 得到了例子.但是他们只是一个一个地实现了那个例子。意味着第一个整数产生然后混淆和程序停止。

这是原始示例:

class ProducerConsumerImpl {
    // producer consumer problem data
    private static final int CAPACITY = 10;
    private final Queue queue = new LinkedList<>();
    private final Random theRandom = new Random();

    // lock and condition variables
    private final Lock aLock = new ReentrantLock();
    private final Condition bufferNotFull = aLock.newCondition();
    private final Condition bufferNotEmpty = aLock.newCondition();

    public void put() throws InterruptedException {
        aLock.lock();
        try {
            while (queue.size() == CAPACITY) {
                System.out.println(Thread.currentThread().getName()
                        + " : Buffer is full, waiting");
                bufferNotEmpty.await();
            }

            int number = theRandom.nextInt();
            boolean isAdded = queue.offer(number);
            if (isAdded) {
                System.out.printf("%s added %d into queue %n", Thread
                        .currentThread().getName(), number);

                // signal consumer thread that, buffer has element now
                System.out.println(Thread.currentThread().getName()
                        + " : Signalling that buffer is no more empty now");
                bufferNotFull.signalAll();
            }
        } finally {
            aLock.unlock();
        }
    }

    public void get() throws InterruptedException {
        aLock.lock();
        try {
            while (queue.size() == 0) {
                System.out.println(Thread.currentThread().getName()
                        + " : Buffer is empty, waiting");
                bufferNotFull.await();
            }

            Integer value = queue.poll();
            if (value != null) {
                System.out.printf("%s consumed %d from queue %n", Thread
                        .currentThread().getName(), value);

                // signal producer thread that, buffer may be empty now
                System.out.println(Thread.currentThread().getName()
                        + " : Signalling that buffer may be empty now");
                bufferNotEmpty.signalAll();
            }

        } finally {
            aLock.unlock();
        }
    }
}

之后我修改了代码并让它像这样工作，前 10 个 Produce，然后 10 个 Consume，循环运行直到程序终止。这是我修改后的代码:

class ProducerConsumerImpl {
    // producer consumer problem data
    private static final int CAPACITY = 10;
    private final Queue<Integer> queue = new LinkedList<>();
    private final Random theRandom = new Random();

    // lock and condition variables
    private final Lock aLock = new ReentrantLock();
    private final Condition bufferNotFull = aLock.newCondition();
    private final Condition bufferNotEmpty = aLock.newCondition();

    public void put() throws InterruptedException {
        aLock.lock();
        try {
            while(true){
                while (queue.size() == CAPACITY) {
                    System.out.println(Thread.currentThread().getName()
                            + " : Buffer is full, waiting");

                    bufferNotEmpty.await();
                }

                int number = theRandom.nextInt();
                boolean isAdded = queue.offer(number);
                if (isAdded) {
                    System.out.printf("%s added %d into queue %n", Thread
                            .currentThread().getName(), number);
                }
               bufferNotFull.signalAll();
            }

        } finally {
            aLock.unlock();
        }
    }

    public void get() throws InterruptedException {
        aLock.lock();
        try {
            while(true){
                while (queue.size() == 0) {
                    System.out.println(Thread.currentThread().getName()
                            + " : Buffer is empty, waiting");
                    bufferNotFull.await();
                }

                Integer value = (Integer)queue.poll();
                if (value != null) {
                    System.out.printf("%s consumed %d from queue %n", Thread
                            .currentThread().getName(), value);
                }
                bufferNotEmpty.signalAll();
            }

        } finally {
            aLock.unlock();
        }
    }
}

修改后它工作正常，生成 10 个随机整数，然后一次又一次地消耗这些整数，直到程序存在/终止。但是因为我不是并发方面的大师/专家。

所以我想问一下我修改的代码有没有问题？

我觉得我的 bufferNotFull.signalAll(); 和 bufferNotEmpty.signalAll(); 放置有问题，因为它每次都会通知等待线程。

如果有问题我该如何解决？或者，如果一切正常，请清除我的 bufferNotFull.signalAll();和 bufferNotEmpty.signalAll();放置，因为每次都可以通知等待线程。 混淆。

注意:只是获取和设置代码。(生产和消费)

修改后的代码输出:

PRODUCER added 1062016967 into queue 
PRODUCER added 1204607478 into queue 
PRODUCER added 1865840177 into queue 
PRODUCER added -1279321362 into queue 
PRODUCER added -190570442 into queue 
PRODUCER added -1344361101 into queue 
PRODUCER added 609239106 into queue 
PRODUCER added -1480451794 into queue 
PRODUCER added 1905208395 into queue 
PRODUCER added -420578734 into queue 
PRODUCER : Buffer is full, waiting
CONSUMER consumed 1062016967 from queue 
CONSUMER consumed 1204607478 from queue 
CONSUMER consumed 1865840177 from queue 
CONSUMER consumed -1279321362 from queue 
CONSUMER consumed -190570442 from queue 
CONSUMER consumed -1344361101 from queue 
CONSUMER consumed 609239106 from queue 
CONSUMER consumed -1480451794 from queue 
CONSUMER consumed 1905208395 from queue 
CONSUMER consumed -420578734 from queue 
CONSUMER : Buffer is empty, waiting
PRODUCER added 1917580670 into queue 
so on.........

已编辑在考虑性能之后，我决定在 signalall(); 语句之前的 put 和 get 方法中添加 if 条件。 (我认为它甚至可以提高性能 0.000000Something)但是中断 await();可能是 deadlock(); 有什么帮助吗？

对于制作人:

if(queue.size() == CAPACITY){
bufferNotFull.signalAll();
}

对于消费者:

if(queue.size() == 0){
bufferNotEmpty.signalAll();
}

Answer 1

修改代码即可。

至于通过在不需要时跳过 .singnalAll() 调用可能的性能优化:仅当条件从 false 更改时发出信号就足够了> 为 true。

对于生产者，您可以使用:

if(isAdded && queue.size() == 1) {
    /*
     * Element has been *actually added* *into empty queue*
     * (previously .size() = 0), thus *queue become non-empty*.
     */
    bufferNotFull.signalAll();
}

对于消费者，您可以使用:

if(value && queue.size() == CAPACITY - 1) {
    /*
     * Element has been *actually consumed* *from full queue*
     * (previousely .size() = CAPACITY), thus *queue become non full*.
     */
    bufferNotEmpty.signalAll();
}

请注意，这样您就不会消除对服务员不必要的通知(例如，消费者只能等待空队列的元素，因此只有添加第一个元素才会唤醒它)。相反，当明确知道没有线程等待它(例如，消费者不能在非空队列上等待)时，您可以消除对 .notifyAll() 的调用。