How do I use the BlockingQueue object?

By Wayan in Concurrency 0 Comment

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.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() – –
Wayan
Latest posts by Wayan (see all)

Leave a Reply

This site uses Akismet to reduce spam. Learn how your comment data is processed.