Java DelayQueue

DelayQueue class is an unbounded blocking queue of delayed elements, in which an element can only be taken when its delay has expired. DelayQueue class is part of java.util.concurrent package.

1. What is delayed element?

• An element will be considered delayed when it implements java.util.concurrent.Delayed interface and it’s getDelay() method return a zero or negative value which indicate that the delay has already elapsed.
• To make things more clear, we can consider that each element stores it activation date/time. As soon as, this timestamp reaches, element is ready to be picked up from queue. The getDelay() method returns the time until the activation of the element.
• Note that an implementation of Delayed interface must define a compareTo() method that provides an ordering consistent with its getDelay() method.
• compareTo(Delayed o) method does not return the actual timestamp, generally. It return a value less than zero if the object that is executing the method has a delay smaller than the object passed as a parameter – otherwise a positive value greater than zero. It will return zero if both the objects have the same delay.

public interface Delayed extends Comparable<Delayed> 
{
	/**
     * Returns the remaining delay associated with this object, in the
     * given time unit.
     *
     * @param unit the time unit
     *
     * @return the remaining delay; zero or negative values indicate
     * that the delay has already elapsed
     */
    long getDelay(TimeUnit unit);
}

2. Delayed Element Example

In given example, we have created an event object. Each event will have event id, name and activation date (after this time it shall be processed).

class DelayedEvent implements Delayed 
{
	private long id;
	private String name;
	private LocalDateTime activationDateTime;

	public DelayedEvent(long id, String name, LocalDateTime activationDateTime) {
		super();
		this.id = id;
		this.name = name;
		this.activationDateTime = activationDateTime;
	}

	public long getId() {
		return id;
	}

	public String getName() {
		return name;
	}

	public LocalDateTime getActivationDateTime() {
		return activationDateTime;
	}

	@Override
	public int compareTo(Delayed that) 
	{
		long result = this.getDelay(TimeUnit.NANOSECONDS) 
						- that.getDelay(TimeUnit.NANOSECONDS);
		if (result < 0) {
			return -1;
		} else if (result > 0) {
			return 1;
		}
		return 0;
	}

	@Override
	public long getDelay(TimeUnit unit) {
		LocalDateTime now = LocalDateTime.now();
		long diff = now.until(activationDateTime, ChronoUnit.MILLIS);
		return unit.convert(diff, TimeUnit.MILLISECONDS);
	}

	@Override
	public String toString() {
		return "DelayedEvent [id=" + id + ", name=" + name + ", activationDateTime=" + activationDateTime + "]";
	}
}

3. What is DelayQueue

• A DelayQueue is an unbounded blocking queue of Delayed elements.
  When a consumer of element wants to take an element from the queue, it can take only when the delay for that particular element has expired.
• DelayQueue is a specialized PriorityQueue that orders elements based on their delay time.
• The head of the queue is the element whose delay expired furthest in the past.
• If there is no element whose delay has expired yet, there is no head element in the queue and poll() will return null.
• Even though unexpired elements cannot be removed using take() or poll(), they are otherwise treated as normal elements in the queue i.e. size() method returns the count of both expired and unexpired elements.
• This queue does not permit null elements because their delay cannot be determined.

4. Java DelayQueue Example

To demo the DelayQueue, I have re-written the producer consumer problem using ScheduledExecutorService. In this program, producer thread is adding events in a DelayQueue. Consumer thread invokes periodically and picks up all items that have expired activation time i.e. in past.

4.1. Event Producer

class DelayedEventProducer implements Runnable 
{
	private final DelayQueue<DelayedEvent> queue;
	private AtomicInteger counter;

	public DelayedEventProducer(DelayQueue<DelayedEvent> queue, AtomicInteger counter) {
		this.queue = queue;
		this.counter = counter;
	}

	@Override
	public void run() 
	{
		LocalDateTime now = LocalDateTime.now();
		int id = counter.incrementAndGet();
		DelayedEvent event = new DelayedEvent(id, "Task-" + id, now);
		System.out.println("Added to queue :: " + event);
		queue.add(event);
	}
}

4.2. Event Consumer

class DelayedEventConsumer implements Runnable 
{
	private final DelayQueue<DelayedEvent> queue;

	public DelayedEventConsumer(DelayQueue<DelayedEvent> queue) {
		this.queue = queue;
	}

	@Override
	public void run() 
	{
		List<DelayedEvent> events = new ArrayList<DelayedEvent>();
		queue.drainTo(events);
		System.out.println("\nEvent processing start **********\n");
		events.stream().forEach(System.out::println);
		System.out.println("\nEvent processing end **********\n");
	}
}

4.3. Main program

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Main 
{
	public static void main(String[] args) throws InterruptedException, ExecutionException 
	{
		DelayQueue<DelayedEvent> queue = new DelayQueue<>();
		AtomicInteger counter = new AtomicInteger();
		ScheduledExecutorService ses = Executors.newScheduledThreadPool(2);
		ses.scheduleAtFixedRate(new DelayedEventProducer(queue, counter), 1, 2, TimeUnit.SECONDS);
		ses.scheduleAtFixedRate(new DelayedEventConsumer(queue), 1, 10, TimeUnit.SECONDS);
	}
}

Program output.

Added to queue :: DelayedEvent [id=1, name=Task-1, activationDateTime=2019-05-27T15:56:33.689]
Added to queue :: DelayedEvent [id=2, name=Task-2, activationDateTime=2019-05-27T15:56:35.619]
Added to queue :: DelayedEvent [id=3, name=Task-3, activationDateTime=2019-05-27T15:56:37.619]
Added to queue :: DelayedEvent [id=4, name=Task-4, activationDateTime=2019-05-27T15:56:39.619]
Added to queue :: DelayedEvent [id=5, name=Task-5, activationDateTime=2019-05-27T15:56:41.619]
Added to queue :: DelayedEvent [id=6, name=Task-6, activationDateTime=2019-05-27T15:56:43.619]

Event processing start **********

DelayedEvent [id=1, name=Task-1, activationDateTime=2019-05-27T15:56:33.689]
DelayedEvent [id=2, name=Task-2, activationDateTime=2019-05-27T15:56:35.619]
DelayedEvent [id=3, name=Task-3, activationDateTime=2019-05-27T15:56:37.619]
DelayedEvent [id=4, name=Task-4, activationDateTime=2019-05-27T15:56:39.619]
DelayedEvent [id=5, name=Task-5, activationDateTime=2019-05-27T15:56:41.619]
DelayedEvent [id=6, name=Task-6, activationDateTime=2019-05-27T15:56:43.619]

Event processing end **********

Added to queue :: DelayedEvent [id=7, name=Task-7, activationDateTime=2019-05-27T15:56:45.620]
Added to queue :: DelayedEvent [id=8, name=Task-8, activationDateTime=2019-05-27T15:56:47.618]
Added to queue :: DelayedEvent [id=9, name=Task-9, activationDateTime=2019-05-27T15:56:49.620]
Added to queue :: DelayedEvent [id=10, name=Task-10, activationDateTime=2019-05-27T15:56:51.618]
Added to queue :: DelayedEvent [id=11, name=Task-11, activationDateTime=2019-05-27T15:56:53.619]

Event processing start **********

DelayedEvent [id=7, name=Task-7, activationDateTime=2019-05-27T15:56:45.620]
DelayedEvent [id=8, name=Task-8, activationDateTime=2019-05-27T15:56:47.618]
DelayedEvent [id=9, name=Task-9, activationDateTime=2019-05-27T15:56:49.620]
DelayedEvent [id=10, name=Task-10, activationDateTime=2019-05-27T15:56:51.618]
DelayedEvent [id=11, name=Task-11, activationDateTime=2019-05-27T15:56:53.619]

Event processing end **********

Clearly, the program is working as intended.

Drop me your questions in comments section related to this Java DelayQueue example program.

Happy Learning !!

Ref : Java docs