How to interrupt a BlockingQueue which is blocking on take()?

Question

I have a class that takes objects from a BlockingQueue and processes them by calling take() in a continuous loop. At some point I know that no more objects will be added to the queue. How do I interrupt the take() method so that it stops blocking?

Here's the class that processes the objects:

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

And here's the method that uses this class to process objects:

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?
}

score 84 · Accepted Answer · answered May 01 '09 at 17:29

84

If interrupting the thread is not an option, another is to place a "marker" or "command" object on the queue that would be recognized as such by MyObjHandler and break out of the loop.

answered May 01 '09 at 17:29

Chris Thornhill

5,321
1
24
21

46

This is also known as the 'Poison Pill Shutdown' approach and is discussed at length in "Java Concurrency in Practice", specifically on pp. 155-156. – Brandon Yarbrough May 01 '09 at 17:52
1

would this approach work if there are multiple consumers? – Rajat Aggarwal Jan 22 '22 at 17:22

score 15 · Answer 2 · edited May 23 '17 at 12:10

Very late but Hope this helps other too as I faced the similar problem and used the poll approach suggested by erickson above with some minor changes,

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

This solved both the problems

Flagged the BlockingQueue so that it knows it has not to wait more for elements
Did not interrupted in between so that processing blocks terminates only when all the items in queue are processed and there are no items remaining to be added

Make `Finished` variable `volatile` to guarantee visibility between threads. See http://stackoverflow.com/a/106787 — lukk, Dec 28 '13 at 03:59
If I'm not mistaken, the final element in the queue will not be processed. When you take the final element from the queue, finished is true and the queue is empty, so it will return before having handled that final element. Add a third condition if (Finished && queue.isEmpty() && obj == null) — Matt R, May 11 '16 at 01:15
@MattR thanks, correctly advised I will edit the answer posted — Deepak Bhatia, May 11 '16 at 10:01

score 15 · Answer 3 · answered May 01 '09 at 17:27

15

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

However, if you do this, the thread might be interrupted while there are still items in the queue, waiting to be processed. You might want to consider using poll instead of take, which will allow the processing thread to timeout and terminate when it has waited for a while with no new input.

answered May 01 '09 at 17:27

erickson

265,237
58
395
493

Yes, it's a problem if the thread is interrupted while there are still items in the queue. To get around this, I added code to make sure the queue is empty before interrupting the thread: while (queue.size()>0) Thread.currentThread().sleep(5000); – MCS May 01 '09 at 17:42
4

@MCS - note for future visitors that your approach here is a hack and should not be reproduced in production code for three reasons. It's always preferred to find an actual way to hook the shutdown. It's never acceptable to use `Thread.sleep()` as an alternative to a proper hook. In other implementations, other threads may be placing things into the queue, and the while loop may never end. – Erick Robertson Apr 21 '14 at 16:15
Thankfully, there's no need to rely on such hacks because one just as easily process it *after* the interrupt if necessary. For example, an "exhaustive" `take()` implementation might look like: `try { return take(); } catch (InterruptedException e) { E o = poll(); if (o == null) throw e; Thread.currentThread().interrupt(); return o; }` However, there's no reason it needs to be implemented at this layer, and implementing it slightly higher up will lead to more efficient code (such as by avoiding the per-element `InterruptedException` and/or by using `BlockingQueue.drainTo()`). – antak Mar 24 '16 at 11:03

score 4 · Answer 4 · answered May 01 '09 at 17:28

4

Interrupt the thread:

thread.interrupt()

answered May 01 '09 at 17:28

stepancheg

4,262
2
33
38

tomasb · Answer 5 · 2012-09-13T02:53:19.467

1

Or don't interrupt, its nasty.

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

when producer puts object to the queue, call queue.notify(), if it ends, call queue.done()
loop while (!queue.isDone() || !queue.isEmpty())
test take() return value for null

edited Sep 13 '12 at 02:53

answered Sep 13 '12 at 02:32

tomasb

1,663
2
22
29

1

I would say that previous solution was a cleaner and a simpler than this – sakthisundar Dec 19 '12 at 14:15
1

The done flag is the same as a poison pill, its just administered differently :) – David Mann Mar 28 '13 at 17:36
2

Cleaner? I doubt it. You don't know when the thread is interrupted exactly. Or let me ask, what is cleaner with that? Its less code, that's a fact. – tomasb Apr 10 '13 at 16:36
You would probably want to make your `done` boolean `volatile`. If you intend to use the class from multiple threads. – Robert May 05 '22 at 12:08

How to interrupt a BlockingQueue which is blocking on take()?

5 Answers5

Linked