Producer is a single main thread and there are n consumers running in an infinite loop all working on blocking queue. how to shutdown

Question

I have a blocking queue on which the producer main thread adds tasks from a file input. I have n consumers that are picking tasks from the blocking queue whenever anyone is free and finds a task in the queue.

Now it might be that my main thread can finish reading all tasks from an input source and pushing into the queue. But i dont want to end my application then and there. Rather, I want to wait for the child consumer processes to finish their tasks after all inputs are exhausted as well as finished by respective child consumers and the consumers are all waiting on the queue for more inputs. Then, I want to proceed to shutdown the application. How can that be done.

All the examples that i have seen are mostly infinitely running producer and consumer simulations for n seconds say. This seems like a practical scenario

Answer 1

Different solutions available (thread.join(), ExecutorService, CountDownLatch.) See How to wait for a number of threads to complete?

Answer 2

The cleanest solution would be to introduce a special task object that makes a consumer terminate. Such a special object is often called poison pill. If there are n consumer threads put n poison pills to the queue when all the real tasks are read from the file. The main thread my end after putting n poison pills to the queue. After each consumer has terminated all threads are finished and the JVM process will terminate.

Another and less elegant solution would be to check if there are still tasks and then signal the consumer threads that this event has happend though a shared boolean variable (called end in the following example). This variable has to be marked volatile in order to make the change from the main thread visible to the consumer threads. Otherwise one must rely on passing other memory barriers such as synchronized. With this solution the consumers may not use take() in order to fetch tasks from the queue because this would block them if no tasks is available (not sure if this could happen in your processing).

import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumer {
    static class Task {} // just something to represent the task

    static volatile boolean end = false; // signal the consumers that they can stop

    public static void main(String[] args) {
        var tasks = new LinkedBlockingQueue<Task>();
        (new Thread(() -> {
            while (!end) {
                var t = tasks.poll();
                if (t != null) {
                    // process t
                }
            }
        })).start(); // Consumer, several of them

        // fill tasks from file...

        // may be started after all tasks are read from the file
        // check every second if there are still tasks in the queue
        while (tasks.size() > 0) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {}
        }
        end = true;
    }
}