Java DelayQueue

Java DelayQueue is a collection and an implementation of BlockingQueue. Unlike ArrayBlockingQueue, DelayQueue is an unbounded collection. The BlockingQueue methods are implemented in such a way that only delay expired elements can be taken out of the queue. If the delay has not expired for any elements in the queue then the poll method will return null.

Element Expiry

Expiry of an element is decided using the getDelay() method of the element. When it returns a value less than or equal to zero that element is considered to be expired.

For this expiry based DelayQueue to work, elements stored should implement the Delayed interface which looks as below,

package java.util.concurrent;

public interface Delayed extends Comparable {

    long getDelay(TimeUnit unit);
}

Delayed interface extends the Comparable interface and so the elements can be compared with one another. They may be used to order the elements in the DelayQueue instance.

Delayed Element Example

package com.javapapers.java.collections;

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

public class DelayElement implements Delayed {
	private String element;
	private long expiryTime;

	public DelayElement(String element, long delay) {
		this.element = element;
		this.expiryTime = System.currentTimeMillis() + delay;
	}

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

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

	@Override
	public String toString() {
		return element + ": " + expiryTime;
	}
}

size() method returns the overall count of elements in the queue which includes the expired elements also. iterator() method of this class does not guarantee traversal of the queue in any particular order.

DelayQueue Producer

Following class a typical producer in a producer-consumer scenario which puts a delay element into a DelayQueue.

package com.javapapers.java.collections;

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

public class DelayQueueProducer implements Runnable {

	protected BlockingQueue<DelayElement> blockingQueue;
	final Random random = new Random();

	public DelayQueueProducer(BlockingQueue<DelayElement> queue) {
		this.blockingQueue = queue;
	}

	@Override
	public void run() {
		while (true) {
			try {
				int delay = random.nextInt(10000);
				DelayElement delayElement = new DelayElement(UUID.randomUUID()
						.toString(), delay);
				System.out.println("Put: "+ delayElement);
				blockingQueue.put(delayElement);
				Thread.sleep(500);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

DelayQueue Consumer

Following class a typical consumer in a producer-consumer scenario which takes elements from a DelayQueue.

package com.javapapers.java.collections;

import java.util.concurrent.BlockingQueue;

public class DelayQueueConsumer implements Runnable {

	protected BlockingQueue<DelayElement> blockingQueue;

	public DelayQueueConsumer(BlockingQueue<DelayElement> queue) {
		this.blockingQueue = queue;
	}

	@Override
	public void run() {
		while (true) {
			try {
				DelayElement delayElement = blockingQueue.take();
				System.out.println(Thread.currentThread().getName()
						+ " take(): " + delayElement);
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

DelayQueue Example

package com.javapapers.java.collections;

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

public class DelayQueueExample {
	public static void main(String[] args) {
		final BlockingQueue<DelayElement> queue = new DelayQueue<DelayElement>();

		DelayQueueProducer queueProducer = new DelayQueueProducer(queue);
		new Thread(queueProducer).start();

		DelayQueueConsumer queueConsumer1 = new DelayQueueConsumer(queue);
		new Thread(queueConsumer1).start();

		DelayQueueConsumer queueConsumer2 = new DelayQueueConsumer(queue);
		new Thread(queueConsumer2).start();

	}
}

DelayQueue Example Output

Put: 24c502c6-40b9-43fa-8f75-12456a8fde61: 1412489289963
Put: 2eef0e9a-cffd-4110-ad53-d1d2afc0d22e: 1412489294635
Put: 110a210c-c7eb-46c8-bb0f-60f83b04024d: 1412489298572
Thread-1 take(): 24c502c6-40b9-43fa-8f75-12456a8fde61: 1412489289963
Put: 120c5cbe-ee74-41f7-82ec-b26e8819e076: 1412489297235
Put: aad3878e-02a7-4adb-ad0b-890860c1a7c8: 1412489294320
Put: a9a20938-b10b-4955-be85-6b9d88b85728: 1412489293823
Put: 218e8793-7a7f-4036-93eb-57ceaba56a53: 1412489299133
Put: 9a90c990-e0c9-4fc5-b461-746527b17387: 1412489299239
Put: 863b3455-48c4-496b-98f9-cfd12a8cafe9: 1412489296604
Put: a97b24c5-ac61-44a3-bdce-f5b88a87bda3: 1412489297165
Put: 6d8e8e3b-d3fa-469e-8efb-220255384787: 1412489297026
Thread-2 take(): a9a20938-b10b-4955-be85-6b9d88b85728: 1412489293823
Put: 9e698e44-d9d7-4aa3-8edc-ada120813a20: 1412489296124