Jak przerwać działanie BlockingQueue, które blokuje się przy pobieraniu()?

Question

Mam klasy, która pobiera obiekty z BlockingQueue i przetwarza je, wywołując take() w pętli ciągłej. W pewnym momencie wiem, że do kolejki nie zostaną dodane żadne obiekty. Jak przerwać metodę take(), aby przestała blokować?

Tutaj jest klasa, która przetwarza 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(); 
    } 
    } 
} 

i tu jest 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? 
} 

Answer 1

przypadku przerwania nici nie jest to możliwe, drugi jest umieszczenie „marker” lub „polecenia” obiekt w kolejce, które są rozpoznawane jako takie przez MyObjHandler i przerwy z pętli.

Answer 2

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 zrobić tego wątku może zostać przerwane, gdy w kolejce znajdują się jeszcze elementy oczekujące na przetworzenie. Możesz rozważyć użycie poll zamiast take, co pozwoli wątkowi przetwarzania na przekroczenie limitu czasu i zakończenie, gdy odczekał chwilę bez nowych danych wejściowych.

Answer 3

przerwań nici:

thread.interrupt() 

Answer 4

Lub nie przerywaj, to jest 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; 
     } 
    } 

1. gdy producent umieszcza obiekt do kolejki, połączenia queue.notify(), jeśli kończy się, połączenia queue.done()
2. pętli while (! Queue.isDone() ||! Queue.isEmpty())
3. Test odbioru() wartość zwracana za zerową

Answer 5

bardzo późno, ale mam nadzieję, że to pomaga też inny jak i w obliczu podobnego problemu i używane poll podejście sugerowane przez erickson above z niewielkimi 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ć oba problemy

1. zasygnalizowała BlockingQueue tak, że wie, że nie musi czekać więcej elementów
2. Nie przerwane pomiędzy, tak że bloki przetwarzania kończą się tylko wtedy, gdy wszystkie elementy w kolejce są przetwarzane i nie ma żadnych elementów, które pozostały do ​​dodania