I have a thread that does two stuff in a loop: wait for objects from a BlockingQueue and then process them. The processing logic has effects that is observable outside the JVM and must be done atomically.
Currently this thread is controlled by a volatile boolean variable like discussed in https://stackoverflow.com/a/10961760/499922. However, this means the thread will not stop if there are no more messages on the queue for it to consume. 'Thread.interrupt()` cannot be used to resolve this because it can interrupt the atomic operation half way.
Is there any way to just interrupt the take() method on the queue but not any other operation?
I've found the best way in instances like this is to use the "Poison Pill" - you place an object on your queue which exists solely for the purpose of toggling a certain action.
It's covered extensively in Java concurrency in practice (page 155, 7.2.3 in my copy). See here for the programmatic example from that section.
'Thread.interrupt()` cannot be used to resolve this because it can interrupt the atomic operation half way.
It doesn't do this. All it does is set a flag which is much the same as setting a volatile field.
Even if you were to use Thread.stop() which is deprecated and can trigger any line of code to throw an error, you can still catch such an error and ensure your data is in a consistent state.
Instead of creating a queue and threads to execute tasks off this queue, I would use an ExecutorService which has both of these and it can be used to cancel/interrupt individual tasks, support multiple threads, shutdown gracefully or immediately and wait for termination.
It is quite likely you would be better off not re-inventing this wheel. ;)
One way to do this is by placing a special "sentinel" object into the queue. The thread would know the meaning of this special object, and will take the appropriate action when it receives it.
Instead of relying on a volatile boolean, rely on the thread's interrupted flag :
while (!Thread.currentThread().isInterrupted()) {
try {
Object value = queue.take();
handle(value);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
This way when the flow is blocked on take() it will exit the loop immediately when the Thread is interrupted. If it is interrupted any other time it will exit the loop after completing its current cycle.
Or you could code it to continue looping after interruption as long as an Object is immediately available on the queue, using poll() from then on.
