Java: Producer = Consumer, how to know when to stop?

Question

I have several workers, that use ArrayBlockingQueue.

Every worker takes one object from queue, process it, and in result can get several objects, that will be put into queue for further processing. So, worker = producer + consumer.

Worker:

public class Worker implements Runnable
{
    private BlockingQueue<String> processQueue = null;

    public Worker(BlockingQueue<String> processQueue)
    {
        this.processQueue = processQueue;
    }

    public void run()
    {
        try
        {
            do
            {
                String item = this.processQueue.take();
                ArrayList<String> resultItems = this.processItem(item);

                for(String resultItem : resultItems)
                {
                    this.processQueue.put(resultItem);
                }
            }
            while(true);
        }
        catch(Exception)
        {
            ...
        }
    }

    private ArrayList<String> processItem(String item) throws Exception
    {
        ...
    }
}

Main:

public class Test
{
    public static void main(String[] args) throws Exception
    {
        new Test().run();
    }

    private void run() throws Exception
    {
        BlockingQueue<String> processQueue = new ArrayBlockingQueue<>(10000);
        processQueue.put("lalala");

        Executor service = Executors.newFixedThreadPool(100);
        for(int i=0; i<100; ++i)
        {
            service.execute(new Worker(processQueue));
        }
    }
}

Whats is the best way to stop workers, when there is no more work ?

First, what I have in mind, is to check periodically how many items in queue and how many items are currently in process. If both are equal to zero, then do something like "shutdownNow()" on ExecutorService. But I am not sure this is the best way.

Answer 1

If there's no more work to do, put a message into the queue saying so and have the workers shut themselves down at their own convenience. This is a good way to prevent data corruption.

If you need to notify another thread that all the workers have gone home, you can use a CountDownLatch to do so.

Answer 2

Sounds like you have your solution--use a separate in-progress queue, the size of which will be the number of items currently being processed. If you use the convention that accesses to either queue is in synchronized(theArrayBlockingQueue) blocks then all should be well. In particular, when moving an item to the processing state, remove it from theArrayBlockingQueue and add it to the processingQueue within the same synchronized block.