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.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.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 receives the marker object from the queue.

package org.kodejava.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();
            }
        }
    }
}

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:

• java.util.concurrent.ArrayBlockingQueue
• java.util.concurrent.DelayQueue
• java.util.concurrent.LinkedBlockingDeque
• java.util.concurrent.LinkedBlockingQueue
• java.util.concurrent.PriorityBlockingQueue
• java.util.concurrent.SynchronousQueue

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.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.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.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()	–	–