У меня есть класс, который берет объекты из 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?
}
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
, что позволит потоку обработки истечь время ожидания и завершиться, когда он некоторое время ждал без нового ввода.источник
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()
).Очень поздно, но надеюсь, что это поможет и другим, поскольку я столкнулся с аналогичной проблемой и использовал
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(); }
Это решило обе проблемы
BlockingQueue
чтобы он знал, что ему больше не нужно ждать элементовисточник
Finished
переменную,volatile
чтобы гарантировать видимость между потоками. См. Stackoverflow.com/a/106787Прервите поток:
источник
Или не перебивай, это мерзко.
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; } }
queue.notify()
, если он заканчивается, вызовqueue.done()
источник
А как насчет
queue.add(new MyObj())
в каком-то производственном потоке, где стоп-флаг сигнализирует потоку-потребителю о завершении цикла while?
источник