Ожидание условия в 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

wait() никогда не уведомляется.

Answer 2

wait () будет продолжаться вечно, потому что вы никогда не вызываете 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);
    }

Как сказал Стив в своем ответе, вы также можете использовать 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

В ваших требованиях указано, что вы собираетесь поместить значения в очередь, когда она пуста - я понимаю, что это несколько значений, а не одно. Если ваши требования немного изменились, чтобы сказать, что вы помещаете один элемент в очередь и ждете, пока он не израсходуется, прежде чем поставить другой, тогда вы будете готовы использовать java.util.concurrent.Exchanger.

Его поведение аналогично BlockingQueue глубины один, но делает это немного больше: он передает объект в обоих направлениях, где каждый участник является и «производителем», и «потребителем». , Следовательно, Exchanger не примет предложение товара «производителем», пока «потребитель» также не будет готов предложить товар обратно. Это не «огонь и забудь» для «продюсера»; сроки производства и потребления взаимосвязаны. Это не позволяет фактическому производителю затопить рабочую очередь потребителя - опять же, как BlockingQueue, - но также останавливает производителя, пока потребитель не завершит последний раунд работы.

В вашем случае у потребителя может не быть ничего полезного для возврата производителю. В любом случае, вы можете сформировать протокол между участниками. Когда производитель хочет отключить потребительский поток, он может предложить нулевое значение. Как только потребитель принимает нулевое значение, производитель может сделать еще один раунд обмена, чтобы завершить запрос на отключение и получить любой конечный результат от потребителя.

Answer 4

Если вы используете 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 пуст.