Код потребителя производителя с ожиданием / уведомлением не работает на втором продукте

Question

Это дополнительный вопрос из моего предыдущего вопроса здесь .

Я использую PriorityBlockingQueue сейчас.Я изменил моего продюсера на следующее:

synchronized(Manager.queue) {
    Manager.queue.add(new Job());
    Manager.queue.notify();
}

И изменил Consumer на следующее.Полный скелет кода здесь:

//my consumer thread run()
public void run() {
synchronized(Manager.queue) {
    while (Manager.queue.peek() == null) {
                System.out.println("111111111111111");
                try {
                    Manager.queue.wait();
                } catch (InterruptedException e) {
                }
            }
    Job job=Manager.queue.peek();
if (job != null) {
                submitJob(job);
                if (job.SubmissionFailed.equals("false")) {
                    // successful submission. Remove from queue. Add to another.
                    Manager.queue.poll();
                    Manager.submissionQueue.put(job.uniqueid, job);
}
}
}

Мой код работает только в первый раз (сначала производят и сначала потребляют), но он не работает во второй раз.Где-то логика ожидания / уведомления не работает, я думаю.Производитель отправляет новые задания в очередь, но потребитель больше не peek.На самом деле, он даже не заходит в цикл while и не печатает 111111111111111.

В чем проблема?Как это исправить?

Answer 1

Вы можете упростить весь этот код до:

В источнике:

Manager.queue.add(new Job());

и в потребителе:

while (true) {
    try {
        submitJob(Manager.queue.take()); //or do something else with the Job
        //your code here, then remove the break
        break;
    } catch (InterruptedException ex) {
        //usually no need to do anything, simply live on unless you
        //caused that
    }
}
//or your code here, then you need an surrounding while and the break

При использовании PriorityBlockingQueue, вам не нужны никакие операторы syncronized, поскольку они уже внутри PriorityBlockingQueue.И согласно документации take() дождитесь добавления элемента в случае необходимости и затем poll s.См. https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/PriorityBlockingQueue.html#take() для справки.

А для InterruptedException вы можете посмотреть здесь: Обработка InterruptedException в Java

Редактировать: добавленоотсутствует try{} catch()