Thread Synchronization with BlockingQueue in Java [closed]

Question

This question is unlikely to help any future visitors; it is only relevant to a small geographic area, a specific moment in time, or an extraordinarily narrow situation that is not generally applicable to the worldwide audience of the internet. For help making this question more broadly applicable, visit the help center.

Closed 9 years ago.

I am new to the concept of a BlockingQueue, and was under the impression that it took out the need for wait() and notify(). I have written the following code as a preliminary test of thread synchronization (omitting some code for clarity):

q = new LinkedBlockingQueue<flight>();

generator = new EventGenerator(q,flight);
southwest = new Airline(q);

new Thread(generator).start();
new Thread(southwest).start();

With the producer class EventGenerator. EventGenerator takes a flight from its initial state (SC) through all of the states until it reaches the gate, AG:

import java.util.concurrent.BlockingQueue;

public class EventGenerator implements Runnable
{
private final BlockingQueue<flight> bqueue;
private final flight f;

EventGenerator(BlockingQueue<flight> q, flight f1)
{
    bqueue = q;
    f = f1;
}

public void run()
{
    try
    {
        while (f.state != "AG") { bqueue.put(produce()); }
    }
    catch (InterruptedException ex)
    {
        System.out.println(ex);
        System.exit(0);
    }
}

flight produce()
{
    if (f.state.equals("SC")){ f.state = "BD"; }
    else if (f.state.equals("BD")) { f.state = "LG"; }
    else if (f.state.equals("LG")) { f.state = "TO"; }
    else if (f.state.equals("TO")) { f.state = "LD"; }
    else if (f.state.equals("LD")) { f.state = "AG"; }

    return f;
}

The consumer class Airline is to receive each change in the state of a flight, and print it:

import java.util.concurrent.BlockingQueue;

public class Airline implements Runnable
{
    private final BlockingQueue<flight> bqueue;

    Airline(BlockingQueue<flight> q)
    {
        bqueue = q;
    }

public void run()
{
    try
    {
        while (!bqueue.isEmpty()) { consume(bqueue.take());} 
    }
    catch (InterruptedException ex)
    {
        System.out.println(ex);
        System.exit(0);
    }
}

void consume(flight f)
{
    System.out.println("Flight no. " + f.flightno + " is now in state " + f.state);
}
}

The project compiles and runs without error. However, I expected (and want) the result to be something to the tune of:

Flight no. 1 is now in state SC
Flight no. 1 is now in state BD
Flight no. 1 is now in state LG
Flight no. 1 is now in state TO
Flight no. 1 is now in state LD
Flight no. 1 is now in state AG

But instead, I almost always get:

Flight no. 1 is now in state AG
Flight no. 1 is now in state AG
Flight no. 1 is now in state AG
Flight no. 1 is now in state AG
Flight no. 1 is now in state AG

Have I implemented the BlockingQueue incorrectly?

Answer 1

The problem you are having at the moment is actually with the produce method. You are changing the state of the same object and placing it on the queue multiple times. What you want to do is actually place a modified copy onto the queue. If your Flight object were immutable, then you wouldn't encounter this issue.

Aside from the above, this answer is also correct about the problems with the conditions in your while loops.

Answer 2

while (f.state != "AG") { bqueue.put(produce()); }

while (!bqueue.isEmpty()) { consume(bqueue.take());}

I suspect both your while conditions are not really what you want: The first one should use .equals() for string comparison. The second assumes that the consumer will execute after the producer (and that the consumer is slower than the producer); Otherwise the consumer will find the queue empty and skip the whole loop.