JAVA中的延时队列DelayQueue

Java中的延时队列DelayQueue是基于优先队列PriorityQueue实现的。

注：PriorityQueue是基于堆（Heap）实现的。堆(Heap)在本质上是一个数组。

Priority queue represented as a balanced binary heap: the two children of queue[n] are queue[2*n+1] and queue[2*(n+1)]. The priority queue is ordered by comparator, or by the elements natural ordering, if comparator is null: For each node n in the heap and each descendant d of n, n <= d. The element with the lowest value is in queue[0], assuming the queue is nonempty.

以下为DelayQueue的使用演示。

1. 定义延时队列的元素

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

public class DelayedItem<T> implements Delayed {

	/**
	 * 到期时间，单位ms
	 */
	private long activeTime;

	private T data;

	public DelayedItem(long activeTime, T data) {
		super();
		this.activeTime = TimeUnit.NANOSECONDS.convert(activeTime, TimeUnit.MILLISECONDS) + System.nanoTime();
		this.data = data;
	}

	public long getActiveTime() {
		return activeTime;
	}

	public T getData() {
		return data;
	}

	@Override
	public int compareTo(Delayed o) {
		long diff = getDelay(TimeUnit.NANOSECONDS) - getDelay(TimeUnit.NANOSECONDS);
		return diff == 0 ? 0 : (diff > 0 ? 1 : -1);
	}

	@Override
	public long getDelay(TimeUnit unit) {
		long diff = unit.convert(activeTime - System.nanoTime(), TimeUnit.NANOSECONDS);
		return diff;
	}
}

2. 定义消费者

import java.util.concurrent.DelayQueue;

public class Consumer implements Runnable {
	
	private DelayQueue<DelayedItem<String>> queue;

	public Consumer(DelayQueue<DelayedItem<String>> queue) {
		super();
		this.queue = queue;
	}

	@Override
	public void run() {
		while (true) {
			try {
				DelayedItem<String> item = queue.take();
				String data = item.getData();
				String time = TimeUtils.getCurTime();
				System.out.printf("[%s] Consuming data:%s
", time, data);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}

3. 实现生产者主线程

import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DelayQueueDemo {

	public static void main(String[] args) {
		DelayQueue<DelayedItem<String>> queue = new DelayQueue<DelayedItem<String>>();
		queue.offer(new DelayedItem<String>(10_000, "A"));
		System.out.printf("[%s] produce message A
", TimeUtils.getCurTime());
		queue.offer(new DelayedItem<String>(20_000, "B"));
		System.out.printf("[%s] produce message B
", TimeUtils.getCurTime());
		
		ExecutorService executorService = Executors.newCachedThreadPool();
		executorService.execute(new Consumer(queue));
		executorService.shutdown();
	}
}

4. 查看输出

[2022-03-22 20:24:37.577] produce message A [2022-03-22 20:24:37.629] produce message B [2022-03-22 20:24:47.311] Consuming data:A [2022-03-22 20:24:57.630] Consuming data:B