Mam klasę, która pobiera obiekty z a BlockingQueue
i przetwarza je, wywołując take()
w ciągłej pętli. W pewnym momencie wiem, że do kolejki nie zostaną dodane żadne obiekty. Jak przerwać take()
metodę, aby przestała blokować?
Oto klasa przetwarzająca obiekty:
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();
}
}
}
A oto metoda, która używa tej klasy do przetwarzania obiektów:
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();
Jeśli jednak to zrobisz, wątek może zostać przerwany, gdy w kolejce nadal znajdują się elementy, które czekają na przetworzenie. Możesz rozważyć użycie
poll
zamiasttake
, co pozwoli wątkowi przetwarzającemu na przekroczenie limitu czasu i zakończenie go, gdy będzie czekał przez chwilę bez nowych danych wejściowych.źródło
Thread.sleep()
jako alternatywy dla właściwego haczyka. W innych implementacjach inne wątki mogą umieszczać rzeczy w kolejce, a pętla while może nigdy się nie kończyć.take()
implementacja może wyglądać następująco:try { return take(); } catch (InterruptedException e) { E o = poll(); if (o == null) throw e; Thread.currentThread().interrupt(); return o; }
Jednak nie ma powodu, aby musiała być implementowana w tej warstwie, a implementacja nieco wyżej doprowadzi do bardziej wydajnego kodu (na przykład przez unikanie elementówInterruptedException
i / lub używającBlockingQueue.drainTo()
).Bardzo późno, ale mam nadzieję, że to pomoże również innym, ponieważ stanąłem przed podobnym problemem i zastosowałem
poll
podejście sugerowane przez Ericksona powyżej z kilkoma drobnymi zmianami,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(); }
To rozwiązało oba problemy
BlockingQueue
aby wiedział, że nie musi dłużej czekać na elementyźródło
Finished
zmiennejvolatile
widoczności gwarancyjnej pomiędzy wątkami. Zobacz stackoverflow.com/a/106787Przerwij wątek:
źródło
Albo nie przerywaj, to paskudne.
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()
, jeśli się kończy, callqueue.done()
źródło
A co z
queue.add(new MyObj())
w jakimś wątku producenta, gdzie flaga stopu sygnalizuje wątkowi konsumenta zakończenie pętli while?
źródło