(Как) я могу "пропустить сигнал" с этим ConcurrentLinkedQueue и sleep ()?

Question

В моем Java-приложении несколько потоков помещают данные в очередь, из которой другой поток (только один) берет объекты и отправляет их.

Иногда поток-потребитель, кажется, не замечает добавления новых элементовв очередь, так как сообщения журнала, указывающие опрос, перестают появляться.Сообщения журнала от производящих потоков указывают, что эти элементы действительно прибывают.Погуглив, некоторые учили меня, что это, кажется, известно как «пропущенный сигнал».Поскольку я не wait не использую и не использую блокировки, я не уверен, относится ли это ко мне.

Что меня действительно удивляет, так это то, что когда я прерываю поток потребителя, он обрабатывает все элементы в очереди.и затем снова хранит молчание, не выходя из него.

Вот основной цикл потока потребителя, потоки производителя выполняют такие вещи, как чтение из сокетов, и их единственное общее свойство - использовать add() в очереди изкоторый потребитель опрашивает.

public class FieldFrontService
{
    // ...

    private ConcurrentLinkedQueue<Transmission> _qRec;

    // ...

    private Runnable _createReceptionist()
    {
        return new Runnable() {
            @Override
            public void run()
            {
                while (!Thread.currentThread().isInterrupted()) {
                    Transmission tx = FieldFrontService.this._qRec.poll();
                    if (null == tx) {
                        try {
                            Thread.sleep(250);
                        } catch (InterruptedException e) {
                            break;
                        }
                    } else {
                        FieldFrontService.this._receiveTransmission(tx);
                    }
                }
            }
        }

    // ...
}

Answer 1

Возможно, вам лучше использовать ExecutorService, который объединяет пул потоков и очередь. И это работает. ;)

public class FieldFrontService {
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    void receiveTransmission(final Transmission tx) {
        executor.execute(new Runnable() {
            public void run() {
                FieldFrontService.this._receiveTransmission(tx);
            }
        });
    }
}

Answer 2

Я думаю, что ваше InterruptedException съедается где-то в вашем стеке вызовов.

Этот шаблон меня беспокоит:

try {                             
    Thread.sleep(250);
} catch (InterruptedException e) {
    break;
} 

Если вы используете подобный шаблон в вашем коде, то состояние прерывания потока было потеряно. Например, если Thread.interrupted() использовалось в вашем стеке вызовов, это сбросит прерванный статус вашего потока . В результате ваш цикл while никогда не завершится: статус был сброшен в false.

Эти две строки вызывают озабоченность:

Transmission tx = FieldFrontService.this._qRec.poll();
// ....
FieldFrontService.this._receiveTransmission(tx); 

Если какой-либо из этих вызовов вызовет Thread.currentThread().interrupted(), вы потеряете логическое значение isInterrupted. Я бы посмотрел реализацию обоих этих методов на случай, если используется исключение / состояние.

Моя обработка InterruptedExceptions отличается. Я предпочитаю отбрасывать прерывание:

try {                             
    Thread.sleep(250);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
} 

Таким образом, я гарантирую, что нет никакой возможности, что мое прерывание было потеряно. По сути, я пытаюсь сказать, нет, серьезно, STOP Thread STOP!

Answer 3

Не потому ли, что оператор break выходит из цикла while, вызывая остановку потока? Что произойдет, если вы удалите оператор break? Или это то, что вы ожидаете?

Я предполагаю, что поток не видит новые элементы, потому что поток уже остановлен из-за исключительной ситуации, которую вы не ожидали.