Синхронизация нити с BlockingQueue в Java

Question

Я новичок в концепции BlockingQueue, и у меня создалось впечатление, что потребовалось wait() и notify(). Я написал следующий код в качестве предварительного теста синхронизации потоков (исключая код для ясности):

q = new LinkedBlockingQueue<flight>(); 

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

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

С классом производителя EventGenerator. EventGenerator берет рейс из своего первоначального состояния (SC) через все состояния, пока она не достигнет ворот, 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; 
} 

Потребительский класс Авиакомпания получить каждое изменение в состоянии полета, и распечатать его :

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

Проект компилируется и запускается без ошибок. Тем не менее, я ожидал (и хотят) результат будет что-то на мотив:

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 

Но вместо этого, я почти всегда получаю:

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 

ли я реализовал BlockingQueue неправильно?

Answer 1

Проблема, которую вы испытываете на данный момент, на самом деле связана с методом produce. Вы меняете состояние одного и того же объекта и ставите его в очередь несколько раз. То, что вы хотите сделать, - это разместить в очереди модифицированную копию. Если ваш объект Flight был неизменным, вы не столкнетесь с этой проблемой.

Помимо вышеуказанного, this answer также относится к проблемам, связанным с условиями в ваших циклах.

Answer 2

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

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

Я подозреваю, что оба ваши while условия не совсем то, что вы хотите: первый следует использовать .equals() для сравнения строк. Второй предполагает, что потребитель выполнит после производителя (и что потребитель медленнее, чем производитель); В противном случае потребитель найдет очередь пустой и пропустит весь цикл.