Tutorial

Java BlockingQueue Example

Published on August 3, 2022
author

Pankaj

Java BlockingQueue Example

Today we will look into Java BlockingQueue. java.util.concurrent.BlockingQueue is a java Queue that support operations that wait for the queue to become non-empty when retrieving and removing an element, and wait for space to become available in the queue when adding an element.

Java BlockingQueue

Java BlockingQueue Java BlockingQueue doesn’t accept null values and throw NullPointerException if you try to store null value in the queue. Java BlockingQueue implementations are thread-safe. All queuing methods are atomic in nature and use internal locks or other forms of concurrency control. Java BlockingQueue interface is part of java collections framework and it’s primarily used for implementing producer consumer problem. We don’t need to worry about waiting for the space to be available for producer or object to be available for consumer in BlockingQueue because it’s handled by implementation classes of BlockingQueue. Java provides several BlockingQueue implementations such as ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue etc. While implementing producer consumer problem in BlockingQueue, we will use ArrayBlockingQueue implementation. Following are some important methods you should know.

  • put(E e): This method is used to insert elements to the queue. If the queue is full, it waits for the space to be available.
  • E take(): This method retrieves and remove the element from the head of the queue. If queue is empty it waits for the element to be available.

Let’s implement producer consumer problem using java BlockingQueue now.

Java BlockingQueue Example - Message

Just a normal java object that will be produced by Producer and added to the queue. You can also call it as payload or queue message.

package com.journaldev.concurrency;

public class Message {
    private String msg;
    
    public Message(String str){
        this.msg=str;
    }

    public String getMsg() {
        return msg;
    }

}

Java BlockingQueue Example - Producer

Producer class that will create messages and put it in the queue.

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

    private BlockingQueue<Message> queue;
    
    public Producer(BlockingQueue<Message> q){
        this.queue=q;
    }
    @Override
    public void run() {
        //produce messages
        for(int i=0; i<100; i++){
            Message msg = new Message(""+i);
            try {
                Thread.sleep(i);
                queue.put(msg);
                System.out.println("Produced "+msg.getMsg());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //adding exit message
        Message msg = new Message("exit");
        try {
            queue.put(msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

Java BlockingQueue Example - Consumer

Consumer class that will process on the messages from the queue and terminates when exit message is received.

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable{

private BlockingQueue<Message> queue;
    
    public Consumer(BlockingQueue<Message> q){
        this.queue=q;
    }

    @Override
    public void run() {
        try{
            Message msg;
            //consuming messages until exit message is received
            while((msg = queue.take()).getMsg() !="exit"){
            Thread.sleep(10);
            System.out.println("Consumed "+msg.getMsg());
            }
        }catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Java BlockingQueue Example - Service

Finally we have to create BlockingQueue service for producer and consumer. This producer consumer service will create the BlockingQueue with fixed size and share with both producers and consumers. This service will start producer and consumer threads and exit.

package com.journaldev.concurrency;

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

public class ProducerConsumerService {

    public static void main(String[] args) {
        //Creating BlockingQueue of size 10
        BlockingQueue<Message> queue = new ArrayBlockingQueue<>(10);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);
        //starting producer to produce messages in queue
        new Thread(producer).start();
        //starting consumer to consume messages from queue
        new Thread(consumer).start();
        System.out.println("Producer and Consumer has been started");
    }

}

Output of the above java BlockingQueue example program is shown below.

Producer and Consumer has been started
Produced 0
Produced 1
Produced 2
Produced 3
Produced 4
Consumed 0
Produced 5
Consumed 1
Produced 6
Produced 7
Consumed 2
Produced 8
...

Java Thread sleep is used in producer and consumer to produce and consume messages with some delay.

Thanks for learning with the DigitalOcean Community. Check out our offerings for compute, storage, networking, and managed databases.

Learn more about our products

About the author(s)

Category:
Tutorial
Tags:

While we believe that this content benefits our community, we have not yet thoroughly reviewed it. If you have any suggestions for improvements, please let us know by clicking the “report an issue“ button at the bottom of the tutorial.

Still looking for an answer?

Ask a questionSearch for more help

Was this helpful?
 
JournalDev
DigitalOcean Employee
DigitalOcean Employee badge
February 12, 2013

I want to read bulk of gmail and store it in my local disk. It takes 20 seconds for reading one mail and save it in the local. I want to reduce it by 5 sec. how can i use the producer consumer logic for this one.

- santhosh

JournalDev
DigitalOcean Employee
DigitalOcean Employee badge
April 9, 2014

Hi Santosh, There is no need to use Producer/Consumer logic in this but still if you want to go ahead, You need to create multiple Producer Thread which will read your mail from Gmail SMTP Server and write it in the BlockingQueue and atleast 4 Consumer Thread.

- HIMANSU NAYAK

    JournalDev
    DigitalOcean Employee
    DigitalOcean Employee badge
    May 18, 2013

    There are two threads. One thread (T1) writes strings in a shared queue (Q) and the other (T2) reads from the queue and prints them. Implement T1 and T2. Note you cannot read from an empty queue and you cannot write to a full queue. You have the following operators CreateEvent #EventName #EventName.Set #EventName.Reset EnterCriticalSection LeaveCrticalSection LOOP END LOOP WAIT #EventName please provide solution

    - akanksha

    JournalDev
    DigitalOcean Employee
    DigitalOcean Employee badge
    April 9, 2014

    Hi Akanksha, What are the purpose of this operator?

    - HIMANSU NAYAK

      JournalDev
      DigitalOcean Employee
      DigitalOcean Employee badge
      June 6, 2013

      very useful example…

      - rp

        JournalDev
        DigitalOcean Employee
        DigitalOcean Employee badge
        July 22, 2013

        Hi, sir if we want to get output like synchronous way like… producer 0 cosumer 0 producer1 consumer1 producer2 cosumer2 . . . .

        - vishal

        JournalDev
        DigitalOcean Employee
        DigitalOcean Employee badge
        July 23, 2013

        in this case, we need to put logic in producer to publish message only when queue is empty.

        - Pankaj

        JournalDev
        DigitalOcean Employee
        DigitalOcean Employee badge
        October 14, 2013

        or we can create a queue of size 1 ?

        - Gaurav

        JournalDev
        DigitalOcean Employee
        DigitalOcean Employee badge
        July 24, 2014

        In this case you can use SynchronousBlockingQueue

        - Emilio

          JournalDev
          DigitalOcean Employee
          DigitalOcean Employee badge
          July 14, 2015

          package com.cisco.onprem.siar.test; public class ProducerConsumerTest { public static void main(String[] args) { CubbyHole c = new CubbyHole(); Producer p1 = new Producer(c, 1); Consumer c1 = new Consumer(c, 1); p1.start(); c1.start(); } } class CubbyHole { private int contents; private boolean available = false; public synchronized int get() { while (available == false) { try { wait(); } catch (InterruptedException e) { } } available = false; notifyAll(); return contents; } public synchronized void put(int value) { while (available == true) { try { wait(); } catch (InterruptedException e) { } } contents = value; available = true; notifyAll(); } } class Consumer extends Thread { private CubbyHole cubbyhole; private int number; public Consumer(CubbyHole c, int number) { cubbyhole = c; this.number = number; } public void run() { int value = 0; for (int i = 0; i < 10; i++) { value = cubbyhole.get(); System.out.println(“Consumer #” + this.number + " got: " + value); } } } class Producer extends Thread { private CubbyHole cubbyhole; private int number; public Producer(CubbyHole c, int number) { cubbyhole = c; this.number = number; } public void run() { for (int i = 0; i < 10; i++) { cubbyhole.put(i); System.out.println(“Producer #” + this.number + " put: " + i); try { sleep((int)(Math.random() * 100)); } catch (InterruptedException e) { } } } }

          - Shiva

            JournalDev
            DigitalOcean Employee
            DigitalOcean Employee badge
            August 23, 2013

            Good explanation…

            - Goutam

              JournalDev
              DigitalOcean Employee
              DigitalOcean Employee badge
              September 7, 2013

              HI Pankaj, Wont sleep be a blocking call. I mean when the producer starts the sleep it will not release the lock on the q that is shared . The consumer will then have to wait to access the queue. Isnt this code blocking …

              - Anshul Gupta

              JournalDev
              DigitalOcean Employee
              DigitalOcean Employee badge
              September 7, 2013

              yes this is blocking code and it’s just used for example purpose. In real life, you will have some logic to produce the message and you will start Producer thread by some other piece of code.

              - Pankaj

                JournalDev
                DigitalOcean Employee
                DigitalOcean Employee badge
                November 29, 2013

                Thanks! very useful example…

                - sudheera

                  JournalDev
                  DigitalOcean Employee
                  DigitalOcean Employee badge
                  July 21, 2014

                  Hello Sir, I am running the same code given by you in above example but its showing given below output: Produced Messagedemo.Message@247cb66a consumednull Not as above given by you.And after adding toString() method in the Consumer class at belwo line it is throwing null pointer exception. System.out.println(“consumed” +msg.getMsg().toString()); Exception in thread “Thread-1” java.lang.NullPointerException at demo.Consumer.run(Consumer.java:24) at java.lang.Thread.run(Unknown Source) Will you please correct me ? Do I need to syncronized them?

                  - Shweta

                  JournalDev
                  DigitalOcean Employee
                  DigitalOcean Employee badge
                  July 21, 2014

                  Without looking at your program, there is no way to know what is missing. For me it’s running fine, checked just now.

                  - Pankaj

                    JournalDev
                    DigitalOcean Employee
                    DigitalOcean Employee badge
                    April 22, 2015

                    i want to create a concurrent asynchronous queue so that the messages or data can be inserted into it using threadpool and another pool retrieves data from the queue and stores it in database (asynchronously) how can i do this if anyone knows about it then please share the code or explain me how to implement it…

                    - Ahmed

                      JournalDev
                      DigitalOcean Employee
                      DigitalOcean Employee badge
                      May 14, 2015

                      Hi Pankaj, Could you kindly put a few more examples where BlockingQueue has been implemented for a larger service e.g. a real-time message queue where it’s being run for almost infinite duration. BTW - This example helps me a lot though! Kindest Regards,

                      - Mohammed Manna

                        JournalDev
                        DigitalOcean Employee
                        DigitalOcean Employee badge
                        June 8, 2016

                        HI, It’s really helpful for the beginner. Thanks, Lalit

                        - Lalit Patil

                        JournalDev
                        DigitalOcean Employee
                        DigitalOcean Employee badge
                        June 8, 2016

                        Thanks Lalit. appreciate your kind words.

                        - Pankaj

                          JournalDev
                          DigitalOcean Employee
                          DigitalOcean Employee badge
                          June 10, 2017

                          Why Consumer is not able to consume everything that Producer has produced. Why it exits without finishing??

                          - Pinesh

                          JournalDev
                          DigitalOcean Employee
                          DigitalOcean Employee badge
                          July 21, 2017

                          I think your main thread died before consumer thread completion . Try join for both threads with main thread .

                          - Rishav

                          JournalDev
                          DigitalOcean Employee
                          DigitalOcean Employee badge
                          February 3, 2018

                          Still its not working .what could be the reason?

                          - Naidu

                            JournalDev
                            DigitalOcean Employee
                            DigitalOcean Employee badge
                            October 27, 2017

                            Hello author, Thank you for tutorial. I’m a C# developer, and trying to build java application for multi platform purposes, i want to ask you a question. how if i only create a producer, to create new object and execute it? i think consumer was unnecessary because we have main thread, and we create 2 other for producer and consumer, sorry if the way i thinking is wrong because i’m just learning blockqueue and create asynchronous from scratch. regards

                            - Vico Ervanda

                              JournalDev
                              DigitalOcean Employee
                              DigitalOcean Employee badge
                              November 3, 2017

                              Thanks Pankaj for the excellent tutorial!

                              - Karthik

                                JournalDev
                                DigitalOcean Employee
                                DigitalOcean Employee badge
                                July 2, 2018

                                Hi, could you change the operator for comparing string from reference to value equality in consumer example ?

                                - Simplicity

                                  JournalDev
                                  DigitalOcean Employee
                                  DigitalOcean Employee badge
                                  September 13, 2018

                                  Using Concurrency Linked List import java.util.Collections; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Scanner; public class Threadexample { public static void main(String[] args) throws InterruptedException { // Object of a class that has both produce() // and consume() methods Scanner in = new Scanner(System.in); int cakes = in.nextInt(); System.out.println(“Number of cakes::”+cakes); final Cake cake = new Cake(); // Create producer thread Thread t1 = new Thread(new Runnable() { @Override public void run() { try { cake.produce(cakes); } catch (InterruptedException e) { e.printStackTrace(); } } }); // Create consumer thread Thread t2 = new Thread(new Runnable() { @Override public void run() { try { cake.consume(cakes); } catch (InterruptedException e) { e.printStackTrace(); } } }); // Start both threads t1.start(); t2.start(); // t1 finishes before t2 t1.join(); t2.join(); } // This class has a list, producer (adds items to list // and consumber (removes items). public static class Cake { // Create a list shared by producer and consumer List list1 = Collections.synchronizedList(new LinkedList()); List list2 = Collections.synchronizedList(new LinkedList()); List list3 = Collections.synchronizedList(new LinkedList()); int capacity = 3; // Function called by producer thread public void produce(int cakes) throws InterruptedException { int value = 0; while (cakes!=0) { synchronized (this) { // producer thread waits while list // is full while (list1.size() == capacity && list2.size() == capacity && list3.size() == capacity) wait(); System.out.println(“Producer produced-” + value); // to insert the jobs in the list if (list1.size() < 3) { System.out.println(“List1 size:” + list1.size()); list1.add(value++); } else if (list2.size() < 3) { System.out.println(“List1 is full:::”); System.out.println(“List2 size:” + list2.size()); list2.add(value++); } else if (list3.size() < 3) { System.out.println(“List2 is full:::”); System.out.println(“List3 size:” + list3.size()); list3.add(value++); } // notifies the consumer thread that // now it can start consuming notify(); // makes the working of program easier // to understand Thread.sleep(1000); cakes–; System.out.println(“Producer cakes:”+cakes); } } } // Function called by consumer thread public void consume(int cakes) throws InterruptedException { while (cakes!=0) { synchronized (this) { // consumer thread waits while list // is empty while (list1.size() == 0 && list2.size() == 0 && list3.size() == 0) wait(); // to retrieve the cake in the list System.out.println(“Consumer cakes:”+cakes); synchronized (list1) { Iterator it = list1.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed-” + it.next()); it.remove(); cakes–; } } synchronized (list2) { Iterator it = list2.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed2-” + it.next()); it.remove(); cakes–; } } synchronized (list3) { Iterator it = list3.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed3-” + it.next()); it.remove(); cakes–; } } // Wake up producer thread notify(); // and sleep Thread.sleep(1000); //cakes–; //System.out.println(“Consumer cakes:”+cakes); } } } } }

                                  - Vinothkumar

                                    JournalDev
                                    DigitalOcean Employee
                                    DigitalOcean Employee badge
                                    January 24, 2019

                                    Hi Pankaj, How in case of multiple threads trying to write in same blocking queue (consider huge count of threads from business perspective view. practically ). Because synchronization would be needed for queue also i think. Please guide in such scenarios. welcome for suggestions

                                    - venkatesh choudhary

                                    JournalDev
                                    DigitalOcean Employee
                                    DigitalOcean Employee badge
                                    August 26, 2020

                                    Since they are all working on the same blockingQueue they all will be considering the same lock hence the implementation of BlockingQueue is sufficient to make sure that the entry is also restricted for all of the producers.

                                    - Manish

                                      JournalDev
                                      DigitalOcean Employee
                                      DigitalOcean Employee badge
                                      June 8, 2016

                                      HI, It’s really helpful for the beginner. Thanks, Lalit

                                      - Lalit Patil

                                      JournalDev
                                      DigitalOcean Employee
                                      DigitalOcean Employee badge
                                      June 8, 2016

                                      Thanks Lalit. appreciate your kind words.

                                      - Pankaj

                                        JournalDev
                                        DigitalOcean Employee
                                        DigitalOcean Employee badge
                                        June 10, 2017

                                        Why Consumer is not able to consume everything that Producer has produced. Why it exits without finishing??

                                        - Pinesh

                                        JournalDev
                                        DigitalOcean Employee
                                        DigitalOcean Employee badge
                                        July 21, 2017

                                        I think your main thread died before consumer thread completion . Try join for both threads with main thread .

                                        - Rishav

                                          JournalDev
                                          DigitalOcean Employee
                                          DigitalOcean Employee badge
                                          October 27, 2017

                                          Hello author, Thank you for tutorial. I’m a C# developer, and trying to build java application for multi platform purposes, i want to ask you a question. how if i only create a producer, to create new object and execute it? i think consumer was unnecessary because we have main thread, and we create 2 other for producer and consumer, sorry if the way i thinking is wrong because i’m just learning blockqueue and create asynchronous from scratch. regards

                                          - Vico Ervanda

                                            JournalDev
                                            DigitalOcean Employee
                                            DigitalOcean Employee badge
                                            November 3, 2017

                                            Thanks Pankaj for the excellent tutorial!

                                            - Karthik

                                              JournalDev
                                              DigitalOcean Employee
                                              DigitalOcean Employee badge
                                              July 2, 2018

                                              Hi, could you change the operator for comparing string from reference to value equality in consumer example ?

                                              - Simplicity

                                                JournalDev
                                                DigitalOcean Employee
                                                DigitalOcean Employee badge
                                                September 13, 2018

                                                Using Concurrency Linked List import java.util.Collections; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Scanner; public class Threadexample { public static void main(String[] args) throws InterruptedException { // Object of a class that has both produce() // and consume() methods Scanner in = new Scanner(System.in); int cakes = in.nextInt(); System.out.println(“Number of cakes::”+cakes); final Cake cake = new Cake(); // Create producer thread Thread t1 = new Thread(new Runnable() { @Override public void run() { try { cake.produce(cakes); } catch (InterruptedException e) { e.printStackTrace(); } } }); // Create consumer thread Thread t2 = new Thread(new Runnable() { @Override public void run() { try { cake.consume(cakes); } catch (InterruptedException e) { e.printStackTrace(); } } }); // Start both threads t1.start(); t2.start(); // t1 finishes before t2 t1.join(); t2.join(); } // This class has a list, producer (adds items to list // and consumber (removes items). public static class Cake { // Create a list shared by producer and consumer List list1 = Collections.synchronizedList(new LinkedList()); List list2 = Collections.synchronizedList(new LinkedList()); List list3 = Collections.synchronizedList(new LinkedList()); int capacity = 3; // Function called by producer thread public void produce(int cakes) throws InterruptedException { int value = 0; while (cakes!=0) { synchronized (this) { // producer thread waits while list // is full while (list1.size() == capacity && list2.size() == capacity && list3.size() == capacity) wait(); System.out.println(“Producer produced-” + value); // to insert the jobs in the list if (list1.size() < 3) { System.out.println(“List1 size:” + list1.size()); list1.add(value++); } else if (list2.size() < 3) { System.out.println(“List1 is full:::”); System.out.println(“List2 size:” + list2.size()); list2.add(value++); } else if (list3.size() < 3) { System.out.println(“List2 is full:::”); System.out.println(“List3 size:” + list3.size()); list3.add(value++); } // notifies the consumer thread that // now it can start consuming notify(); // makes the working of program easier // to understand Thread.sleep(1000); cakes–; System.out.println(“Producer cakes:”+cakes); } } } // Function called by consumer thread public void consume(int cakes) throws InterruptedException { while (cakes!=0) { synchronized (this) { // consumer thread waits while list // is empty while (list1.size() == 0 && list2.size() == 0 && list3.size() == 0) wait(); // to retrieve the cake in the list System.out.println(“Consumer cakes:”+cakes); synchronized (list1) { Iterator it = list1.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed-” + it.next()); it.remove(); cakes–; } } synchronized (list2) { Iterator it = list2.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed2-” + it.next()); it.remove(); cakes–; } } synchronized (list3) { Iterator it = list3.iterator(); while (it.hasNext()) { System.out.println(“Consumer consumed3-” + it.next()); it.remove(); cakes–; } } // Wake up producer thread notify(); // and sleep Thread.sleep(1000); //cakes–; //System.out.println(“Consumer cakes:”+cakes); } } } } }

                                                - Vinothkumar

                                                  JournalDev
                                                  DigitalOcean Employee
                                                  DigitalOcean Employee badge
                                                  January 24, 2019

                                                  Hi Pankaj, How in case of multiple threads trying to write in same blocking queue (consider huge count of threads from business perspective view. practically ). Because synchronization would be needed for queue also i think. Please guide in such scenarios. welcome for suggestions

                                                  - venkatesh choudhary

                                                  JournalDev
                                                  DigitalOcean Employee
                                                  DigitalOcean Employee badge
                                                  August 26, 2020

                                                  Since they are all working on the same blockingQueue they all will be considering the same lock hence the implementation of BlockingQueue is sufficient to make sure that the entry is also restricted for all of the producers.

                                                  - Manish

                                                    Try DigitalOcean for free

                                                    Click below to sign up and get $200 of credit to try our products over 60 days!

                                                    Sign up

                                                    Join the Tech Talk
                                                    Success! Thank you! Please check your email for further details.

                                                    Please complete your information!

                                                    Become a contributor for community

                                                    Get paid to write technical tutorials and select a tech-focused charity to receive a matching donation.

                                                    DigitalOcean Documentation

                                                    Full documentation for every DigitalOcean product.

                                                    Resources for startups and SMBs

                                                    The Wave has everything you need to know about building a business, from raising funding to marketing your product.

                                                    Get our newsletter

                                                    Stay up to date by signing up for DigitalOcean’s Infrastructure as a Newsletter.

                                                    New accounts only. By submitting your email you agree to our Privacy Policy

                                                    The developer cloud

                                                    Scale up as you grow — whether you're running one virtual machine or ten thousand.

                                                    Get started for free

                                                    Sign up and get $200 in credit for your first 60 days with DigitalOcean.*

                                                    *This promotional offer applies to new accounts only.