Java中的多线程等待条件

Question

我想让一个线程在队列变空时将值放入队列，并在队列变空时等待这个条件。这是我试图使用的代码，但它会打印出来

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

所以它只工作了一次。有什么问题吗

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

只要使用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 2

wait（）将永远持续，因为您从未调用notify（）

您可以在队列中等待，并在希望等待的线程唤醒时调用notify。为此，请将Producer改为：

    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改为：

    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();
    }
}