Building Cloud Expertise with centron - Our Tutorials

Whether you are a beginner or an experienced professional, our practical tutorials provide you with the knowledge you need to make the most of our cloud services.

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.

Important Methods

  • 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.

Java BlockingQueue Examples

Message

        package com.journaldev.concurrency;

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

            public String getMsg() {
                return msg;
            }
        }

Producer

        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();
                }
            }
        }

Consumer

        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();
                }
            }
        }

Service

        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

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.

Starten Sie Ihre Cloud-Reise mit unserer kostenlosen Trial-Version!

Entdecken Sie die grenzenlosen Möglichkeiten unserer Cloud-Dienste ganz unverbindlich. Melden Sie sich jetzt für unsere kostenlose Trial-Version an und erleben Sie, wie unsere innovativen Lösungen Ihr Business transformieren können.

Try for free!