Wie unterbreche ich eine BlockingQueue, die bei take () blockiert?

82

Ich habe eine Klasse, die Objekte von a nimmt BlockingQueueund sie verarbeitet, indem sie take()in einer Endlosschleife aufruft . Irgendwann weiß ich, dass der Warteschlange keine Objekte mehr hinzugefügt werden. Wie unterbreche ich die take()Methode, damit sie nicht mehr blockiert?

Hier ist die Klasse, die die Objekte verarbeitet:

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();
    }
  }
}

Und hier ist die Methode, mit der diese Klasse Objekte verarbeitet:

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
quelle

Antworten:

70

Wenn das Unterbrechen des Threads keine Option ist, können Sie auch ein "Marker" - oder "Befehl" -Objekt in die Warteschlange stellen, das von MyObjHandler als solches erkannt wird, und aus der Schleife ausbrechen.

Chris Thornhill
quelle
39
Dies wird auch als "Poison Pill Shutdown" -Ansatz bezeichnet und wird ausführlich in "Java Concurrency in Practice", insbesondere auf den Seiten 155-156, erörtert.
Brandon Yarbrough
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();

In diesem Fall wird der Thread möglicherweise unterbrochen, solange sich noch Elemente in der Warteschlange befinden, die auf die Verarbeitung warten. Möglicherweise möchten Sie pollstatt verwenden take, wodurch das Zeitlimit für den Verarbeitungsthread überschritten und beendet wird, wenn er eine Weile ohne neue Eingabe gewartet hat.

erickson
quelle
Ja, es ist ein Problem, wenn der Thread unterbrochen wird, während sich noch Elemente in der Warteschlange befinden. Um dies zu umgehen, habe ich Code hinzugefügt, um sicherzustellen, dass die Warteschlange leer ist, bevor der Thread unterbrochen wird: <code> while (queue.size ()> 0) Thread.currentThread (). Sleep (5000); </ code>
MCS
3
@MCS - Beachten Sie für zukünftige Besucher, dass Ihr Ansatz hier ein Hack ist und aus drei Gründen nicht im Produktionscode reproduziert werden sollte. Es wird immer bevorzugt, einen tatsächlichen Weg zu finden, um das Herunterfahren zu aktivieren. Es ist niemals akzeptabel, Thread.sleep()als Alternative zu einem richtigen Haken zu verwenden. In anderen Implementierungen stellen andere Threads möglicherweise Dinge in die Warteschlange, und die while-Schleife endet möglicherweise nie.
Erick Robertson
Zum Glück besteht keine Notwendigkeit, sich auf solche Hacks zu verlassen, da man sie bei Bedarf genauso einfach nach dem Interrupt verarbeiten kann. Eine "erschöpfende" take()Implementierung könnte beispielsweise so aussehen: Es try { return take(); } catch (InterruptedException e) { E o = poll(); if (o == null) throw e; Thread.currentThread().interrupt(); return o; } gibt jedoch keinen Grund, warum sie auf dieser Ebene implementiert werden muss, und eine etwas höhere Implementierung führt zu effizienterem Code (z. B. durch Vermeiden des Elements pro Element InterruptedExceptionund / oder mit BlockingQueue.drainTo()).
Antak
13

Sehr spät, aber ich hoffe, das hilft auch anderen, da ich mich dem ähnlichen Problem gestellt und den pollvon erickson oben vorgeschlagenen Ansatz mit einigen geringfügigen Änderungen verwendet habe.

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();
}

Dies löste beide Probleme

  1. Markiert das, BlockingQueuedamit es weiß, dass es nicht mehr auf Elemente warten muss
  2. Wurde dazwischen nicht unterbrochen, sodass die Verarbeitungsblöcke nur beendet werden, wenn alle Elemente in der Warteschlange verarbeitet wurden und keine Elemente mehr hinzugefügt werden müssen
dbw
quelle
2
Machen Sie eine FinishedVariable volatile, um die Sichtbarkeit zwischen den Threads zu gewährleisten. Siehe stackoverflow.com/a/106787
lukk
2
Wenn ich mich nicht irre, wird das letzte Element in der Warteschlange nicht verarbeitet. Wenn Sie das letzte Element aus der Warteschlange nehmen, ist beendet wahr und die Warteschlange ist leer, sodass sie zurückgegeben wird, bevor das letzte Element behandelt wurde. Fügen Sie eine dritte Bedingung hinzu, wenn (Fertig && queue.isEmpty () && obj == null)
Matt R
@ MattR danke, richtig geraten, ich werde die Antwort bearbeiten
dbw
1

Unterbrechen Sie den Thread:

thread.interrupt()
stepancheg
quelle
0

Oder nicht unterbrechen, es ist böse.

    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. Wenn der Produzent ein Objekt in die Warteschlange stellt, rufen Sie auf queue.notify(), wenn es endet, rufen Sie aufqueue.done()
  2. Schleife while (! queue.isDone () ||! queue.isEmpty ())
  3. test take () Rückgabewert für null
tomasb
quelle
1
Ich würde sagen, dass die vorherige Lösung sauberer und einfacher war als diese
sakthisundar
1
Die fertige Flagge ist die gleiche wie eine Giftpille, sie wird nur anders verabreicht :)
David Mann
Reiniger? Ich bezweifle das. Sie wissen nicht genau, wann der Thread unterbrochen wird. Oder lassen Sie mich fragen, was ist damit sauberer? Es ist weniger Code, das ist eine Tatsache.
Thomasb