Java中等待条件

Question

我想创建一个线程，当队列变空时将值放入队列中，并在队列不空时等待这一条件。这是我尝试使用的代码，但它打印出

Adding new
Taking Value 1
Taking Value 2
Taking Value 3
Taking Value 4

So it is only work only 一次。问题是什么？

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


public class SO {
    public String test;
    public String[] list = new String[] {test};

    public static void main(String[] args) {
        new SO();
    }

    public SO() {
        go();
    }

    BlockingQueue<String> qq = new LinkedBlockingQueue<String>();

    class Producer implements Runnable {
        public void run() {
            try {
                while (true) {
                    synchronized (this) {
                        while (qq.size() > 0)
                            wait();

                        System.out.println("Adding new");
                        qq.put("Value 1");
                        qq.put("Value 2");
                        qq.put("Value 3");
                        qq.put("Value 4");
                    }
                }
            } catch (InterruptedException ex) {}
        }
    }

    class Consumer implements Runnable {
        public void run() {
            try {
                while(true) {
                    System.out.println("Taking " + qq.take()+". "+String.valueOf(qq.size())+" left");
                    Thread.sleep(1000);
                }
            } catch (InterruptedException ex) {}
        }
    }

    public void go() {
        Producer p = new Producer();
        Consumer c = new Consumer();

        new Thread(p).start();
        new Thread(c).start();
    }
}

Answer 1

wait() 永远不会被通知。

Answer 2

wait() 将永远持续下去，因为您从未调用过 notification()。

当您希望唤醒等待线程时，您可以等待队列并调用通知。为此，您可以将 Producer 更改为 read:

    synchronized (qq) {
        while (qq.size() > 0)
            qq.wait();

            System.out.println("Adding new");
            qq.put("Value 1");
            qq.put("Value 2");
            qq.put("Value 3");
            qq.put("Value 4");
    }

并将 Consumer 更改为 read:

    while(true) {
        synchronized (qq) {
            System.out.println("Taking " + qq.take() + ". " + String.valueOf(qq.size()) + " left");
            qq.notify();
        }
        Thread.sleep(1000);
    }

正如 Steve 在他的回答中所说，您还可以在消费者线程中使用 wait() ，这样它就可以等待，直到列表中有内容而不是休眠。所以你的代码将变成：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class SO {
    public String test;
    public String[] list = new String[] { test };

    public static void main(String[] args) {
        new SO();
    }

    public SO() {
        go();
    }

    BlockingQueue qq = new LinkedBlockingQueue();

    class Producer implements Runnable {
        public void run() {
            try {
                while (true) {
                    synchronized (qq) {
                        if (!qq.isEmpty()) {
                            qq.wait();
                        }

                        System.out.println("Adding new");
                        qq.put("Value 1");
                        qq.put("Value 2");
                        qq.put("Value 3");
                        qq.put("Value 4");
                        qq.notify();
                    }
                }
            } catch (InterruptedException ex) {
            }
        }
    }

    class Consumer implements Runnable {
        public void run() {
            try {
                while (true) {
                    synchronized (qq) {
                        System.out.println("Taking " + qq.take() + ". "
                                + String.valueOf(qq.size()) + " left");
                        if (qq.isEmpty()) {
                            qq.notify();
                            qq.wait();
                        }
                    }
                }
            } catch (InterruptedException ex) {
            }
        }
    }

    public void go() {
        Producer p = new Producer();
        Consumer c = new Consumer();

        new Thread(p).start();
        new Thread(c).start();
    }
}

Answer 3

既然您使用的是 BlockingQueue，则不必使用 synchronized，因为 BlockingQueue 默认情况下是同步的。如果你想使用同步，那么你应该通过同一个对象进行同步：

 synchronized(theSameObjectInstance) {
    while (true) {                        
        while (qq.size() > 0)
            theSameObjectInstance.wait();                            

        System.out.println("Adding new");
        qq.put("Value 1");
        ...

        theSameObjectInstance.notifyAll();
    }
 }

并且消费者的方法应该包装在 synchronized(theSameObjectInstance) 中以便接收通知，消费者还应该在某个地方“等待”，例如当 qq 时是空的。

Answer 4

您的要求表明您打算在队列为空时将值放入队列中 - 我认为是多个值，而不仅仅是一个。如果您的需求稍有变化，即您将一个项目放入队列中，并等到它被消耗后再放入另一个项目，那么您就可以使用java.util.concurrent.Exchanger。

它的行为类似于 BlockingQueue 的深度为一，但它做了更多的事情：它双向传递一个对象，其中每个参与者既是“生产者”又是“消费者”。因此，在“消费者”也准备好返还商品之前，交换者不会接受“生产者”提供的商品。对于“制作人”来说，这不是“一劳永逸”；而是“一劳永逸”。生产和消费时间是环环相扣的。这可以防止实际的生产者淹没消费者的工作队列——同样，像 BlockingQueue 一样——但它也会阻止生产者直到消费者完成最后一轮工作。

在您的情况下，消费者可能没有任何有用的东西可以返回给生产者。无论如何，您可以在参与者之间制定协议。当生产者希望消费者线程关闭时，它可以提供一个空值。一旦消费者接受空值，生产者可能会再进行一轮交换以完成关闭请求并从消费者收集任何最终结果。