Как прервать BlockingQueue, который блокирует take ()?

82

У меня есть класс, который берет объекты из a BlockingQueueи обрабатывает их, вызывая take()в непрерывном цикле. В какой-то момент я знаю, что больше никаких объектов в очередь не будет. Как мне прервать take()метод, чтобы он перестал блокироваться?

Вот класс, обрабатывающий объекты:

public class MyObjHandler implements Runnable {

  private final BlockingQueue<MyObj> queue;

  public class MyObjHandler(BlockingQueue queue) {
    this.queue = queue;
  }

  public void run() {
    try {
      while (true) {
        MyObj obj = queue.take();
        // process obj here
        // ...
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }
}

А вот метод, который использует этот класс для обработки объектов:

public void testHandler() {

  BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100);  

  MyObjectHandler  handler = new MyObjectHandler(queue);
  new Thread(handler).start();

  // get objects for handler to process
  for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext(); ) {
    queue.put(i.next());
  }

  // what code should go here to tell the handler
  // to stop waiting for more objects?
}
MCS
источник

Ответы:

72

Если прерывание потока невозможно, другой вариант - поместить объект «маркер» или «команду» в очередь, который будет распознаваться MyObjHandler как таковой и выйти из цикла.

Крис Торнхилл
источник
40
Это также известно как подход «выключения Poison Pill» и подробно обсуждается в «Java Concurrency in Practice», в частности на стр. 155-156.
Брэндон Ярбро,
14
BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100);
MyObjectHandler handler = new MyObjectHandler(queue);
Thread thread = new Thread(handler);
thread.start();
for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext(); ) {
  queue.put(i.next());
}
thread.interrupt();

Однако, если вы сделаете это, поток может быть прерван, пока в очереди есть элементы, ожидающие обработки. Возможно, вы захотите рассмотреть возможность использования pollвместо take, что позволит потоку обработки истечь время ожидания и завершиться, когда он некоторое время ждал без нового ввода.

Эриксон
источник
Да, это проблема, если поток прерывается, когда в очереди еще есть элементы. Чтобы обойти это, я добавил код, чтобы убедиться, что очередь пуста, прежде чем прерывать поток: <code> while (queue.size ()> 0) Thread.currentThread (). Sleep (5000); </code>
MCS
3
@MCS - обратите внимание для будущих посетителей, что ваш подход здесь является взломом и не должен воспроизводиться в производственном коде по трем причинам. Всегда предпочтительно найти реальный способ отключить выключение. Никогда нельзя использовать Thread.sleep()в качестве альтернативы правильному крючку. В других реализациях другие потоки могут помещать вещи в очередь, и цикл while может никогда не закончиться.
Эрик Робертсон
К счастью, нет необходимости полагаться на такие хаки, потому что их так же легко обработать после прерывания, если это необходимо. Например, «исчерпывающая» take()реализация может выглядеть так: try { return take(); } catch (InterruptedException e) { E o = poll(); if (o == null) throw e; Thread.currentThread().interrupt(); return o; } Однако нет причин, по которым ее нужно реализовывать на этом уровне, и реализация ее немного выше приведет к более эффективному коду (например, за счет отказа от отдельных элементов InterruptedExceptionи / или с помощью BlockingQueue.drainTo()).
antak
13

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

class MyObjHandler implements Runnable 
{
    private final BlockingQueue<MyObj> queue;
    public volatile boolean Finished;  //VOLATILE GUARANTEES UPDATED VALUE VISIBLE TO ALL
    public MyObjHandler(BlockingQueue queue) 
    {
        this.queue = queue;
        Finished = false;
    }
    @Override
    public void run() 
    {        
        while (true) 
        {
            try 
            {
                MyObj obj = queue.poll(100, TimeUnit.MILLISECONDS);
                if(obj!= null)//Checking if job is to be processed then processing it first and then checking for return
                {
                    // process obj here
                    // ...
                }
                if(Finished && queue.isEmpty())
                    return;

            } 
            catch (InterruptedException e) 
            {                   
                return;
            }
        }
    }
}

public void testHandler() 
{
    BlockingQueue<MyObj> queue = new ArrayBlockingQueue<MyObj>(100); 

    MyObjHandler  handler = new MyObjHandler(queue);
    new Thread(handler).start();

    // get objects for handler to process
    for (Iterator<MyObj> i = getMyObjIterator(); i.hasNext(); )
    {
        queue.put(i.next());
    }

    // what code should go here to tell the handler to stop waiting for more objects?
    handler.Finished = true; //THIS TELLS HIM
    //If you need you can wait for the termination otherwise remove join
    myThread.join();
}

Это решило обе проблемы

  1. Отметил BlockingQueue чтобы он знал, что ему больше не нужно ждать элементов
  2. Не прерывались между ними, поэтому блоки обработки завершаются только тогда, когда все элементы в очереди обработаны, и не осталось элементов, которые нужно добавить
dbw
источник
2
Сделайте Finishedпеременную, volatileчтобы гарантировать видимость между потоками. См. Stackoverflow.com/a/106787
lukk
2
Если не ошибаюсь, последний элемент в очереди обрабатываться не будет. Когда вы берете последний элемент из очереди, завершение истинно, а очередь пуста, поэтому он вернется до того, как обработает этот последний элемент. Добавьте третье условие if (Finished && queue.isEmpty () && obj == null)
Matt R
@MattR спасибо, правильно посоветовал, я отредактирую опубликованный ответ
dbw
1

Прервите поток:

thread.interrupt()
Степанчег
источник
0

Или не перебивай, это мерзко.

    public class MyQueue<T> extends ArrayBlockingQueue<T> {

        private static final long serialVersionUID = 1L;
        private boolean done = false;

        public ParserQueue(int capacity) {  super(capacity); }

        public void done() { done = true; }

        public boolean isDone() { return done; }

        /**
         * May return null if producer ends the production after consumer 
         * has entered the element-await state.
         */
        public T take() throws InterruptedException {
            T el;
            while ((el = super.poll()) == null && !done) {
                synchronized (this) {
                    wait();
                }
            }

            return el;
        }
    }
  1. когда производитель помещает объект в очередь, вызов queue.notify(), если он заканчивается, вызовqueue.done()
  2. цикл while (! queue.isDone () ||! queue.isEmpty ())
  3. test take () возвращаемое значение для нуля
томасб
источник
1
Я бы сказал, что предыдущее решение было чище и проще, чем это
sakthisundar
1
Флаг готовности такой же, как у отравленной пилюли, только вводится по-другому :)
Дэвид Манн,
Очиститель? Я в этом сомневаюсь. Вы не знаете, когда именно поток прерывается. Или позвольте мне спросить, что от этого чище? Меньше кода, это факт.
tomasb
0

А как насчет

queue.add(new MyObj())

в каком-то производственном потоке, где стоп-флаг сигнализирует потоку-потребителю о завершении цикла while?

Сэм Гинрич
источник
1
Прежде чем отвечать на старый вопрос с принятым ответом (ищите зеленый ✓), а также на другие ответы, убедитесь, что ваш ответ добавляет что-то новое или является полезным по отношению к ним. Таким образом, ответ на ваш вопрос дается принятым ответом на вопрос OP. - Формат вопросов и ответов Stack Overflow не предназначен для обсуждения на форуме. Не отвечайте, спрашивая "как насчет того, чтобы сделать X?" поскольку он не отвечает на вопрос OP, а скорее является комментарием. См. Правила для авторов .
Иво Мори
... видно, моё сообщение может быть удалено
Сэм Гинрич