Java 等待 - 在 while 循环内/外同步

Question

我只是在处理消费者/生产者问题，以便熟悉 Java 中的并发问题。我的问题可能与 C/P 问题本身相邻。在下面的代码 sn-ps 中，如果我使用版本 1，程序似乎运行良好，而使用版本 2 我似乎在一段时间后出现死锁。

我只会发布 Producer 类，因为不需要 Consumer 类。

• 缓冲区是生产者存放其输出的地方
• 我只是在其中添加 1 的整数
• while(true) 是保持线程连续运行

我的问题是：我知道在 Java API 中调用 wait() 的结构是：同步 -> while -> 等待。但是在这种情况下会发生什么：while -> synchronized -> wait？如果在 wait() 期间，Consumer 调用 notifyAll() 并且 Producer 再次唤醒，那么代码不会从调用 wait() 的地方继续，最终它会到达 while(buffer.remainingCapacity() == 0) ?

我已经完成了四处挖掘，看看之前是否有人问过这个问题，但找不到任何具体的内容。

版本 1

public class Producer extends Thread {
    ArrayBlockingQueue<Integer> buffer = new ArrayBlockingQueue<>(10);
    public Producer(ArrayBlockingQueue<Integer> buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        while (true) {
            synchronized(buffer) {
                while (buffer.remainingCapacity() == 0) {
                    try {
                        buffer.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            try {
                synchronized (buffer) {
                    buffer.add(1);
                    System.out.println("Producer active with remaining capacity: " + buffer.remainingCapacity());
                    buffer.notifyAll();
                }
            } catch (IllegalStateException ex) {
                ex.printStackTrace();
            }
        }
    }
}

第 2 版

public class Producer extends Thread {
    ArrayBlockingQueue<Integer> buffer = new ArrayBlockingQueue<>(10);

    public Producer(ArrayBlockingQueue<Integer> buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        while (true) {
            while (buffer.remainingCapacity() == 0) {
                synchronized (buffer) {
                    try {
                        buffer.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            try {
                synchronized (buffer) {
                    buffer.add(1);
                    System.out.println("Producer active with remaining capacity: " + buffer.remainingCapacity());
                    buffer.notifyAll();
                }
            } catch (IllegalStateException ex) {
                ex.printStackTrace();
            }
        }
    }
}

只是为了它，这是另一个版本代码，似乎可以解决问题。我只是把它放在这里，以防它对其他人有帮助，或者如果有人可以告诉我它是否错误以及为什么。与上面的版本相比，当涉及到它们的位置时，代码中的 wait 和 notifyAll 方法“翻转”了。

while(true) {
            while(buffer.remainingCapacity() == 0){
                synchronized (buffer){
                    buffer.notifyAll();
                }
            }
            try {
                buffer.add(1);
                System.out.println("Producer active with remaining capacity: " + buffer.remainingCapacity());
            } catch (IllegalStateException ex) {
                synchronized (buffer){
                    try {
                        buffer.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

Answer 1

同步要考虑的事情是“什么操作需要是原子的？”在您的示例中，检查然后等待必须是原子的。否则，请考虑以下事件序列：

• 线程 1：已执行检查但失败
• 线程 2：更新缓冲区更改检查结果
• 线程 2：已调用通知
• 线程 1：等待调用

在这种情况下，Thread1 错过了通知并阻塞，直到另一个通知到来。将检查放在同步块中消除了这种可能性，因为 Thread2 在 Thread1 调用 wait 之前无法调用 notify。

Answer 2

首先，针对生产者/消费者问题使用ArrayBlockingQueue 或任何BlockingQueue 的正确方法显示在该界面的documentation 中，针对您的示例稍作调整：

制片人

    try {
        while (true) {
            queue.put(1);
        }
    } catch (InterruptedException ex) {
        // something want's us to stop.
    }

消费者

    try {
        while (true) {
            Integer thing = queue.take();
        }
    } catch (InterruptedException ex) {
        // something want's us to stop.
    }

恰好有 0 个synchronized/wait/notify，因为这正是这些队列在put 和take 中提供的。你应该在实践中使用这些。自己写只会导致错误。 由于您手动使用synchronized/wait/notify，因此您可以将任何东西用作缓冲区。例如。一个ArrayList。当然，它们上没有 remainingCapacity() 方法，所以绑定队列毕竟不是一个坏选择。

但是除了死锁之外，您的实现还有另一个问题：如果有超过 1 个生产者或消费者，它将失败：您的代码每次迭代有 2 个同步块。现在可能发生的是：

• 队列正好有 1 个插槽打开。
• 生产者 1，检查队列中是否有空间，保持同步
• 生产者 2，检查队列中是否有空间，保持同步
• 生产者 2，进入第二个同步，放入一些东西，队列现在已满，离开
• 生产者 1，进入 2nd 同步，放东西.. arr - 崩溃。

作为 1 个大型原子操作，您真正需要同步的是整个检查然后等待并放置的事情。你最终会得到粗略的

    try {
        while (true) {
            // produce outside of synchronized
            Integer product = 1;

            // BEGIN put atomic
            synchronized (buffer) {
                while (buffer.remainingCapacity() == 0) {
                    buffer.wait();
                }
                buffer.add(product);
                buffer.notifyAll();
                System.out.println("Producer active with remaining capacity: "
                        + buffer.remainingCapacity());
            }
            // END put atomic

            // consumer should likewise consume the product outside
            // of the synchronized block
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

BEGIN和END之间的部分也大致是put()在ArrayBlockingQueue中的实现。

PS：您的“翻转”示例也确实是错误的。除了这两个同步块： 当你有一个缓慢的消费者时，看看你的程序的 CPU 消耗：它会因为生产者积极地卡在循环通知中而达到峰值。这基本上是 100% 的 CPU 负载。

此外，使用IllegalStateException 来检测您不应该添加一些东西，然后重试至少是不好的风格，我不确定当涉及到多个生产者或消费者时是否没有另一个隐藏的死锁.

最后，您依靠buffer.add() 内置的同步功能来确保您甚至得到该异常。在编写自己的 p/c 时，切勿在 synchronized 之外读取或写入缓冲区。

毕竟，您可能希望使用像ArrayList 这样的非同步集合来实现这一点，因此您的代码中的错误不会被其他东西的同步所隐藏。