Producer Consumer Using PriorityBlockingQueue Example in Java

Producer Consumer Using PriorityBlockingQueue Example in Java

java.util.concurrent.PriorityBlockingQueue class is an implementation of BlockingQueue. In our earlier example of Producer Consumer in Java using BlockingQueue we understood how ArrayBlockingQueue class is used.

The general difference between the two is that :
1) ArrayBlockingQueue accepts any java object
2) PriorityBlockingQueue accepts class objects that implements the java.lang.Comparable interface

So, the elements from the PriorityBlockingQueue are read in particular order. The ordering is done by the priority part.

One more thing is, iterator of PriorityBlockingQueue does not guarantee a priority order.

Example Implementation: Below class PersonComparable implements a Comparable interface and performs sorting based on Age.

1) Adding 3 PersonComparable objects with Age = 65, 55, 45 in the Producer Thread
2) Removing 3 PersonComparable objects using take() method. Printing the age of 3 PersonComparable object will display ordering in sorted order by Age i.e. 35, 55, 65

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;

public class PersonComparable implements Comparable<PersonComparable> {

 private int age;

 public int getAge() {
  return age;
 }

 public void setAge(int age) {
  this.age = age;
 }

 @Override
 public int compareTo(PersonComparable personComparable) {
  int compareAge = ((PersonComparable) personComparable).getAge();
  return this.age - compareAge;
 }

 public static void main(String args[]) {
  final PersonComparable person = new PersonComparable();
  person.setAge(65);
  
  final PersonComparable personOne = new PersonComparable();
  personOne.setAge(55);

  final PersonComparable personTwo = new PersonComparable();
  personTwo.setAge(45);

  final BlockingQueue<PersonComparable> queue = new PriorityBlockingQueue<PersonComparable>();

  new Thread(new Runnable() {
   @Override
   public void run() {
    try {
     queue.put(person); // Age 65
     queue.put(personOne); // Age 55
     queue.put(personTwo); // Age 45
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
  }, "Producer").start();

  new Thread(new Runnable() {
   @Override
   public void run() {
    try {
     System.out.println(queue.take().getAge()); // Age 45 Sorted
     System.out.println(queue.take().getAge()); // Age 55 Sorted
     System.out.println(queue.take().getAge()); // Age 65 Sorted
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
  }, "Consumer").start();
 }
}