Как немедленно отпустить потоки, ожидающие на BlockingQueue

Question

Рассмотрите BlockingQueue и несколько потоков, ожидающих на poll(long, TimeUnit) (возможно, также на take()).

Теперь очередь пуста, и необходимо уведомить ожидающие потоки, что они могут перестать ждать. Ожидаемое поведение должно быть возвращено либо null, либо объявлено InterruptedException.

Object.notify() не будет работать для LinkedBlockingQueue, поскольку потоки ждут от внутреннего замка.

Любой простой способ?

Answer 1

Обычным способом является прерывание потоков, но это, конечно же, требует правильной обработки прерываний.

Это означает, что улавливать и обрабатывать InterruptedException s надлежащим образом вокруг методов блокировки, а также регулярно проверять (и действовать) флаг interrupted.

В спецификации API или языка нет ничего, что связывает прерывание с какой-либо определенной семантикой отмены, но на практике использование прерывания для чего угодно, кроме отмены, является хрупким и трудно выдерживается в больших приложениях. [...]

Прерывание - это, как правило, самый разумный способ осуществить отмену.

Говорит Java Concurrency in Practice в разделе 7.1.1. Пример обработки прерывания правильно, от этого же (это производитель нить, а не потребитель, но разница незначительна в текущем контексте):

class PrimeProducer extends Thread { 
    private final BlockingQueue<BigInteger> queue; 

    PrimeProducer(BlockingQueue<BigInteger> queue) { 
     this.queue = queue; 
    } 

    public void run() { 
     try { 
      BigInteger p = BigInteger.ONE; 
      while (!Thread.currentThread().isInterrupted()) 
       queue.put(p = p.nextProbablePrime()); 
     } catch (InterruptedException consumed) { 
      /* Allow thread to exit */ 
     } 
    } 
    public void cancel() { interrupt(); } 
} 

Альтернативным решением было бы установить параметр времени ожидания poll достаточно низкий, так что поток просыпается регулярно и может заметить прерывания достаточно быстро. Тем не менее, я считаю, что всегда хорошая практика обрабатывать InterruptedException явно в соответствии с вашей конкретной политикой аннулирования потока.

Answer 2

Я бы сказал, что с вашим дизайном что-то не так. Нити, потребляющие BlockingQueue, не должны прерываться таким образом. Если они должны делать что-то еще с регулярным интервалом (например, проверять состояние переменной), а также потреблять из очереди, то вы должны использовать метод poll() с установленным таким образом тайм-аутом таким образом, чтобы эти два действия могли чередоваться.

Answer 3

Javadoc для BlockingQueue предлагает хороший способ:

BlockingQueue не свойственно поддержки любого рода «закрыть» или «выключение» операции, чтобы указать, что не больше пунктов будут добавлены. Потребности и использование таких функций, как правило, зависит от реализации. Например, Обычная тактика для производителей: Вставить специальный конец потока или яд объектов, которые толкованы соответственно, если их принять потребители.