如何中断阻塞 take()的 BlockingQueue？

我有一个类，它从 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?
}
44425 次浏览
小开 
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.

小开

Interrupt the thread:

thread.interrupt()
小开
最佳答案

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.

小开

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;
}
}
  1. when producer puts object to the queue, call queue.notify(), if it ends, call queue.done()
  2. loop while (!queue.isDone() || !queue.isEmpty())
  3. test take() return value for null
小开

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

  1. Flagged the BlockingQueue so that it knows it has not to wait more for elements
  2. 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

提交答案