Producer consumer code with wait/notify doesn'

Question

This is a follow-up question from my previous question asked here.

I am using a PriorityBlockingQueue now. I changed my producer to the following:

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

And changed Consumer to the following. Full code skeleton is here:

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

My code only works for the first time (first produce and first consume), but it doesn't work for the second time. Somewhere the wait/notify logic fails I guess. The producer pushes new jobs to the queue, but the consumer doesn't peek any more items. In fact, it doesn't even go to the while loop and no more 111111111111111 printing.

What is the problem? How to fix it?

Answer 1

You could simplify all this code to just:

In the Producer:

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

and in the Consumer:

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

When using an PriorityBlockingQueue, you don't need any syncronized statements, since they're inside the PriorityBlockingQueue already. And according to the documentation take() wait for an element to be added if necessary and than polls it. See https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/PriorityBlockingQueue.html#take() for reference.

And for the InterruptedException you might want to take a look here: Handling InterruptedException in Java

Edit: added missing try{} catch()