How to Stop a Multi-Threaded Consumer Safely Using a Blocking Queue?

Question

What is the best way to properly stop a multi-threaded consumer when using an ArrayBlockingQueue in Java?

// Sample code for the ConsumerWorker class
public class ConsumerWorker implements Runnable {
   private BlockingQueue<Produced> inputQueue;
   private final static Produced POISON = new Produced(-1);
   
   public ConsumerWorker(BlockingQueue<Produced> inputQueue) {
       this.inputQueue = inputQueue;
   }

   @Override
   public void run() {
       while (true) {
           try {
               Produced queueElement = inputQueue.take();
               if(queueElement == POISON) {
                   break;
               }
               // Process the queueElement
           } catch (InterruptedException e) {
               Thread.currentThread().interrupt();
           }
       }
   }
   
   public void stopRunning() throws InterruptedException {
       inputQueue.put(POISON);
   }
}

Answer

To properly stop a multi-threaded consumer in Java that processes tasks from an ArrayBlockingQueue, you can use the 'poison pill' strategy. This ensures that all workers know when to stop processing without leaving tasks unprocessed.

// Example code for a safe stop using poison pills:
public class ConsumerWorker implements Runnable {
    private BlockingQueue<Produced> inputQueue;
    private final static Produced POISON = new Produced(-1);
    
    public ConsumerWorker(BlockingQueue<Produced> inputQueue) {
        this.inputQueue = inputQueue;
    }
    
    @Override
    public void run() {
        while (true) {
            try {
                Produced queueElement = inputQueue.take();
                if(queueElement == POISON) {
                    break; // Gracefully exit loop
                }
                // Process the queueElement
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt(); // Restore interrupted state
            }
        }
    }
    
    public void stopRunning() throws InterruptedException {
        inputQueue.put(POISON); // Adds the poison pill to the queue
    }
}

Causes

Race conditions can occur when signaling consumers to stop before all tasks are completed.
Using a simple boolean flag does not guarantee safe communication between the producer and consumer.

Solutions

Implement a producer-consumer pattern utilizing a 'poison pill' to signal consumers to stop when all tasks are complete.
Use the BlockingQueue's built-in mechanisms for thread-safe signaling of state changes.

Common Mistakes

Mistake: Relying on checking queue size combined with a running flag for safe exiting.

Solution: Use a poison pill to ensure clarity when stopping consumers.

Mistake: Not handling InterruptedException correctly in consumer threads.

Solution: Always restore the interrupted state by calling Thread.currentThread().interrupt() in catch block.

Helpers

BlockingQueue
Multi-threaded consumer
Producer-consumer pattern
Java concurrency
Poison pill strategy