How do I close or shutdown a BlockingQueue?

In this example you’ll learn how to close or shutdown a BlockingQueue when no more element available in the queue. We will use the common strategy by making the Producer to send a marker object in a Producer – Consumer scenario. This marker object also known as the poison object will be considered as a sign that the queue contain no more object that need to be processed. Which then will allow us to break the operation of the consumer thread.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueShutdown {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(32);

        MyDataProducer producer = new MyDataProducer(queue);
        MyDataConsumer consumer = new MyDataConsumer(queue);

        new Thread(producer).start();
        new Thread(consumer).start();
    }
}

Below is the Producer object that put data into the queue. The string DONE is our marker object. This is the last data will be placed in the queue for the consumer to pick up.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.BlockingQueue;

public class MyDataProducer implements Runnable {
    BlockingQueue<String> queue;

    public MyDataProducer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("MyDataProducer.run");
        String[] data = {"D001", "D002", "D003", "D004", "D005", "DONE"};

        try {
            for (String element : data) {
                queue.put(element);
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

The Consumer object loops to retrieve elements from the queue. And it will break the loop and ended the thread when it retrieve the marker object from the queue.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.BlockingQueue;

public class MyDataConsumer implements Runnable {
    BlockingQueue<String> queue;

    public MyDataConsumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("MyDataConsumer.run");

        while (true) {
            try {
                String element = queue.take();
                if ("DONE".equals(element)) {
                    System.out.println("Exiting consumer thread, " +
                            "end of data reached.");
                    break;
                }
                System.out.println("Element = " + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

How do I use the ArrayBlockingQueue?

ArrayBlockingQueue is one implementation of the java.util.concurrent.BlockingQueue which internally store the queue elements inside an array. The ArrayBlockingQueue can store elements for the size defined when the object is initialized, by the constructor. Once it size is defined it cannot be change or resize.

The code snippet below demonstrate the ArrayBlockingQueue class. We initialize the queue to allow its internal array to store maximum of 64 elements.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ArrayBlockingQueueExample {
    private BlockingQueue<String> sharedQueue = new ArrayBlockingQueue<>(64);

    public static void main(String[] args) {
        new ArrayBlockingQueueExample().createProducerConsumer();
    }

    private void createProducerConsumer() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    System.out.println(Thread.currentThread().getName());
                    try {
                        sharedQueue.put("DATA");
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, "Producer Thread").start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    System.out.print(Thread.currentThread().getName() + "=> ");
                    try {
                        System.out.println(sharedQueue.take());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, "Consumer Thread-1").start();
    }
}

How do I use the BlockingQueue object?

A java.util.concurrent.BlockingQueue is an interface that extends the java.util.Queue that add special support for blocking operations. It will wait or block for the queue to become available when retrieving element operations occurs and wait or block the storing element operations until the queue has space available.

There are some implementations available for the java.util.concurrent.BlockingQueue interface. These implementations include the following classes:

Below is an example of how to use the BlockingQueue. In the example we use the ArrayBlockingQueue implementation of the interface. This example then create different thread for each Producer and Consumer object. Both of these threads use a shared blocking queue where the Producer object store some elements and the Consumer object try to retrieve the elements.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(32);

        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        new Thread(producer, "Producer").start();
        new Thread(consumer, "Consumer").start();
    }
}

Here is the Producer class. We define an array of string and use a for-loop to iterate this array to store the element into the queue. We do this by calling the put() method of the BlockingQueue. The put() method block the process if there is no space available in the queue. The calling of the Thread.sleep() method here causes the Consumer to block while waiting for the object available in the queue.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {
    BlockingQueue<String> queue;

    Producer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("Producer.run");
        String[] data = {"D001", "D002", "D003", "D004", "D005"};

        try {
            for (String element : data) {
                queue.put(element);
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

The Consumer object loops to retrieve element from the queue. In this example we retrieve an element using the take() method of the BlockingQueue and print it out into the console.

package org.kodejava.example.util.concurrent;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable {
    BlockingQueue<String> queue;

    Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("Consumer.run");

        while (true) {
            System.out.println("Reading queue...");

            try {
                System.out.println(queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

The BlockingQueue has four different methods set for the storing and retrieving elements in the queue. Each of this method have different behavior.

Throws exception Special value Blocks Times out
Insert add(e) offer(e) put(e) offer(e, time, unit)
Remove remove() poll() take() poll(time, unit)
Examine element() peek()