java.util.concurrent.DelayQueue Example

Ashraf SarhanOctober 17th, 2014Last Updated: October 17th, 2014

1 226 4 minutes read

In this example we shall show you how to make use of DelayQueue class. DelayQueue is an unbounded time-based scheduling BlockingQueue of Delayed elements backed by a heap where an element can only be taken when its delay has expired. If multiple elements have expired delays, the element with the longest delay expiration will be taken first.

DelayQueue characteristics:

DelayQueue is an unbounded (unlimited size) queue.
Element from DelayQueue can only be taken when its delay has expired and If no delay has expired there is no head and poll will return null.
At the head of the queue , element with furthest expired delay time is found.
DelayQueue doesn’t permit Null element.
In DelayQueue, Object class that inside the queue should implement the Delayed interface. This interface forces you to implement 2 methods:
- getDelay: a method that returns how much time is left before the delay completes, getDelay method is important because Java decided to dequeue element from queue if getDelay() method returns a value less than or equal to zero.”
- compareTo: The Delayed interface extends the Comparable interface, so Delayed implementations must override the compareTo() to specify how they should be ordered with respect to other Delayed objects.

There are many cases where we can use the DelayQueue like fax server which cannot lose any incoming fax requests while the server is currently transmitting because it can place all incoming fax requests in a queue, returning immediately to the client requesting the transmission. Also, there are some social web sites which leaves the users comments editable which can be placed in DelayQueue with a specific delay time, after expiration those comments will be taken from DelayQueue to save them and become not editable.

Let’s see our example below which illustrate the DelayQueue usage.

Example:

DelayObject.java:

In the implementation class DelayObject of the Delayed interface, we have to implement the getDelay and the compareTo methods.

package com.jcg;

        import java.util.concurrent.Delayed;
        import java.util.concurrent.TimeUnit;

/**
 * @author ashraf
 */
public class DelayObject implements Delayed {
    private String data;
    private long startTime;

    public DelayObject(String data, long delay) {
        this.data = data;
        this.startTime = System.currentTimeMillis() + delay;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = startTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        if (this.startTime < ((DelayObject) o).startTime) {
            return -1;
        }
        if (this.startTime > ((DelayObject) o).startTime) {
            return 1;
        }
        return 0;
    }

    @Override
    public String toString() {
        return "{" +
                "data='" + data + '\'' +
                ", startTime=" + startTime +
                '}';
    }
}

DelayQueueProducer.java:

In DelayQueueProducer, we are creating a DelayObject object with its attributes and pushed it to our DelayQueue.

package com.jcg;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;

/**
 * @author ashraf
 *
 */
public class DelayQueueProducer {
	
    // Creates an instance of blocking queue using the DelayQueue.
    private BlockingQueue queue;

	private final Random random = new Random();
	
	public DelayQueueProducer(BlockingQueue queue) {
		super();
		this.queue = queue;
	}
	
	private Thread producerThread = new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                try {
                  
                    // Put some Delayed object into the DelayQueue.
                    int delay = random.nextInt(10000);
                    DelayObject object = new DelayObject(
                            UUID.randomUUID().toString(), delay);

                    System.out.printf("Put object = %s%n", object);
                    queue.put(object);
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }, "Producer Thread");
	
	public void start(){
		this.producerThread.start();
	}

}

DelayQueueConsumer.java:

In DelayQueueConsumer, it tries to find the elements that has expired and takes them from the queue, if it could not, it waits until an element will be put and expired.

package com.jcg;

import java.util.concurrent.BlockingQueue;

/**
 * @author ashraf
 *
 */
public class DelayQueueConsumer {
	
	private String name;
	
    private BlockingQueue queue;

	public DelayQueueConsumer(String name, BlockingQueue queue) {
		super();
		this.name = name;
		this.queue = queue;
	}
	
	private Thread consumerThread =  new Thread(new Runnable() {
        @Override
        public void run() {
            while (true) {
                try {
                    // Take elements out from the DelayQueue object.
                    DelayObject object = queue.take();
                    System.out.printf("[%s] - Take object = %s%n",
                            Thread.currentThread().getName(), object);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    });
	
	public void start(){
		this.consumerThread.setName(name);
		this.consumerThread.start();
	}

}

DelayQueueTest.java:

Finally, we are running the DelayQueueTest where we are creating a new DelayQueue of DelayObject and starting a new DelayQueueProducer then we let our new DelayQueueConsumer to take and print the expired elements.

package com.jcg;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;

/**
 * @author ashraf
 * 
 */
public class DelayQueueTest {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		
		// Creates an instance of blocking queue using the DelayQueue.
		BlockingQueue queue = new DelayQueue();
		
		// Starting DelayQueue Producer to push some delayed objects to the queue 
		new DelayQueueProducer(queue).start();
		
		// Starting DelayQueue Consumer to take the expired delayed objects from the queue
		new DelayQueueConsumer("Consumer Thread-1", queue).start();

	}

}

Output:

Put object = {data='3cde8aa3-0b0b-468d-8e59-9068644586d1', startTime=1413541344594}
Put object = {data='682ebf05-b938-4533-9cc5-3a88ad889ade', startTime=1413541340627}
[Consumer Thread-1] - Take object = {data='933ff3a1-ee8f-4e1c-8398-33a81c77f06f', startTime=1413541339483}
Put object = {data='7ed4ecdc-7030-4dce-9314-a32902ce72eb', startTime=1413541341068}
Put object = {data='23464a45-e523-4de5-99ca-dbf2b9a80cf2', startTime=1413541342212}
[Consumer Thread-1] - Take object = {data='682ebf05-b938-4533-9cc5-3a88ad889ade', startTime=1413541340627}
Put object = {data='ac6586c4-ec0d-47ad-b248-625204ba62be', startTime=1413541345148}
Put object = {data='7a77bfa4-16c0-4a81-a217-2b28910a6af3', startTime=1413541350509}
[Consumer Thread-1] - Take object = {data='7ed4ecdc-7030-4dce-9314-a32902ce72eb', startTime=1413541341068}
Put object = {data='c310baf2-492c-46ee-b5cc-531a076b62a9', startTime=1413541350364}
Put object = {data='9dd1886e-94e0-4d08-a993-3bb60bcb92cf', startTime=1413541344156}
[Consumer Thread-1] - Take object = {data='37669690-a687-4168-9f04-725839c7ae59', startTime=1413541341072}
Put object = {data='d05ff5f3-ad6c-4301-a31c-59701ba3c2a3', startTime=1413541348100}
Put object = {data='5668690c-902f-46fa-b147-b9060bcd1ab5', startTime=1413541347165}
[Consumer Thread-1] - Take object = {data='f5238ef0-be1e-472a-acd3-974f90a490b9', startTime=1413541342086}