take（）でブロックしているBlockingQueueを中断する方法は？

StackOverflow https://stackoverflow.com/questions/812342

質問

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?
}
役に立ちましたか？

解決

スレッドの中断がオプションではない場合、別の方法として、「マーカー」を配置します。または＆quot;コマンド＆quot; MyObjHandlerによってそのように認識され、ループから抜け出すキュー上のオブジェクト。

他のヒント

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

ただし、これを行うと、キューにアイテムが残っている間にスレッドが中断され、処理されるのを待機する場合があります。 take の代わりに poll を使用します。新しい入力はありません。

非常に遅いが、これが他の人にも役立つことを願っています同様の問題に直面し、poll アプローチを使用しました/ questions / 812342 / how-to-interrupt-a-blockingqueue-which-is-blocking-on-take / 812354＃812354 ">上記のエリクソンに若干の変更を加え、

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. 間に割り込みがなかったため、キュー内のすべてのアイテムが処理され、追加するアイテムが残っていない場合にのみ処理ブロックが終了します

スレッドを中断します：

thread.interrupt()

または中断しないでください、その厄介です。

    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. を呼び出します
  2. ループwhile（！queue.isDone（）||！queue.isEmpty（））
  3. nullのtake（）戻り値のテスト
ライセンス： CC-BY-SA と 帰属
所属していません StackOverflow
scroll top