延迟阻塞队列DelayQueue

川长思鸟来 2022-07-14 11:46 201阅读 0赞

**延迟阻塞队列DelayQueue**

根据JDK文档描述：

*Delayed元素的一个**无界阻塞队列**，只有在**延迟期满时才能从中提取元素**。该队列的**头部**是**延迟期满**后保存时间**最长**的Delayed元素。如果**延迟都还没有期满**，**则队列没有头部**，并且poll将返回null。当一个元素的 getDelay(TimeUnit.NANOSECONDS)方法返回一个**小于等于0的值**时，将**发生到期**。即使无法使用take或poll移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size方法同时返回到期和未到期元素的计数。此队列**不允许使用null元素**。*

因此DelayQueue可以理解为一个使用时间作为比较条件的PriorityBlockingQueue(优先级阻塞队列)。

举个例子：

夏天买来一批食品放入冷藏室,每种食品在冷藏室都有一个保存时间，超过该时间就会变质。食品检查员经常检查食品,超过冷藏时间的食品就要拿出来扔掉。

首先，定义一个食品类：

import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    public class Food implements Delayed {
    	private String foodName;
    	private long saveTime;// 保存时间
    	private long expireTime;// 过期时刻=当前时间+保存时间
    
    	public Food(String foodName, long saveTime) {
    		this.foodName = foodName;
    		this.saveTime = saveTime;
    		this.expireTime = TimeUnit.NANOSECONDS.convert(saveTime, TimeUnit.SECONDS) + System.nanoTime();
    	}
    
    	public int compareTo(Delayed o) {
    		Food that = (Food) o;
    		if (this.expireTime > that.expireTime) {// 过期时刻越靠后，越排在队尾.
    			return 1;
    		} else if (this.expireTime == that.expireTime) {
    			return 0;
    		} else {
    			return -1;
    		}
    	}
    
    	public long getDelay(TimeUnit unit) {
    		return unit.convert(this.expireTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    	}
    
    	public String getFoodName() {
    		return this.foodName;
    	}
    
    	public long getExpireTime() {
    		return this.expireTime;
    	}
    
    	public long getSaveTime() {
    		return this.saveTime;
    	}
    
    }

这里，食品类实现了Delayed接口，因此重写了compareTo和getDelay方法。getDelay返回与此对象相关的剩余延迟时间，以给定的时间单位表示。

下面定义一个食品检查员的类：

import java.util.concurrent.DelayQueue;
    
    public class FoodChecker implements Runnable {
    	private DelayQueue<Food> queue;
    
    	public FoodChecker(DelayQueue<Food> queue) {
    		this.queue = queue;
    	}
    
    	public void run() {
    		try {
    			System.out.println("开始检查!");
    			boolean flag = true;
    			while (flag) {
    				Food food = queue.take();// 此处会阻塞,没有时过期食品时不会取出
    				System.out.println(food.getFoodName() + "食品过期!保存时间:" + food.getSaveTime() + "天.");
    				if (queue.isEmpty()) {
    					flag = false;
    				}
    			}
    			System.out.println("检查完毕!");
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }

食品检查员持有一个DelayQueue，并且不断从该队列中取出过期食品处理掉。

最后，入口函数如下：

import java.util.Random;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    import delay.demo.Food;
    import delay.demo.FoodChecker;
    
    public class DelayQueueTest {
    
    	public static void main(String[] args) {
    		DelayQueue<Food> queue = new DelayQueue<Food>();
    		Random r = new Random();
    		queue.add(new Food("A", getRandomDay(r)));
    		queue.add(new Food("B", getRandomDay(r)));
    		queue.add(new Food("C", getRandomDay(r)));
    		queue.add(new Food("D", getRandomDay(r)));
    		queue.add(new Food("E", getRandomDay(r)));
    		queue.add(new Food("F", getRandomDay(r)));
    
    		ExecutorService es = Executors.newSingleThreadExecutor();
    		es.execute(new FoodChecker(queue));
    		 es.shutdown();
    	}
    
    	private static long getRandomDay(Random r) {
    		long nextLong = r.nextInt(20);
    		System.out.println("nextLong:" + nextLong);
    		return nextLong;
    	}
    }

在队列中放入几种食品，并随机给一个保存时间。由于FoodChecker只有一个线程，所以这里使用的是SingleThreadExecutor。

运行结果如下：

nextLong:14
    nextLong:2
    nextLong:11
    nextLong:7
    nextLong:7
    nextLong:14
    开始检查!
    B食品过期!保存时间:2天.
    D食品过期!保存时间:7天.
    E食品过期!保存时间:7天.
    C食品过期!保存时间:11天.
    A食品过期!保存时间:14天.
    F食品过期!保存时间:14天.
    检查完毕!

这个例子中使用的是DelayQueue的take方法，该方法获取并移除此队列的头部，在可从此队列获得延迟到期的元素之前会一直等待(即阻塞)。另外，还有几个常用方法如下：

**poll()**：获取并移除此队列的头，如果此队列不包含具有已到期延迟时间的元素，则**返回 null**。(非阻塞)

**peek()**：获取但不移除此队列的头部；如果此队列为空，则返回 null。与poll不同，如果队列中**没有到期元素**可用，则此方法**返回下一个将到期的元素**（如果存在一个这样的元素）。

附：另外一篇不错的文章，一个关于缓存的例子

http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html