龙空技术网

java延迟队列DelayQueue使用及原理

架构师的修炼之路 805

前言:

眼前你们对“java使用队列”大体比较注意,大家都想要知道一些“java使用队列”的相关文章。那么小编同时在网上网罗了一些关于“java使用队列””的相关文章,希望咱们能喜欢,姐妹们一起来了解一下吧!

概述

  java延迟队列提供了在指定时间才能获取队列元素的功能,队列头元素是最接近过期的元素。没有过期元素的话,使用poll()方法会返回null值,超时判定是通过getDelay(TimeUnit.NANOSECONDS)方法的返回值小于等于0来判断。延时队列不能存放空元素。

延时队列实现了Iterator接口,但iterator()遍历顺序不保证是元素的实际存放顺序。

队列元素

  DelayQueue<E extends Delayed>的队列元素需要实现Delayed接口,该接口类定义如下:

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);

}

  由Delayed定义可以得知,队列元素需要实现getDelay(TimeUnit unit)方法和compareTo(Delayed o)方法, getDelay定义了剩余到期时间,compareTo方法定义了元素排序规则,注意,元素的排序规则影响了元素的获取顺序,将在后面说明。

内部存储结构  

DelayedQuene的元素存储交由优先级队列存放。

public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {    private final transient ReentrantLock lock = new ReentrantLock();    private final PriorityQueue<E> q = new PriorityQueue<E>();//元素存放

DelayedQuene的优先级队列使用的排序方式是队列元素的compareTo方法,优先级队列存放顺序是从小到大的,所以队列元素的compareTo方法影响了队列的出队顺序。

若compareTo方法定义不当,会造成延时高的元素在队头,延时低的元素无法出队。

获取队列元素  非阻塞获取

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

E first = q.peek();

if (first == null || first.getDelay(NANOSECONDS) > 0)

return null;

else

return q.poll();

} finally {

lock.unlock();

}

}------------------------------------------------------------------------------------------------------------------------  PriorityQueue队列peek()方法。

public E peek() { return (size == 0) ? null : (E) queue[0];}

  由代码我们可以看出,获取元素时,总是判断PriorityQueue队列的队首元素是否到期,若未到期,返回null,所以compareTo()的方法实现不当的话,会造成队首元素未到期,当队列中有到期元素却获取不到的情况。因此,队列元素的compareTo方法实现需要注意。

  阻塞方式获取

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

for (;;) {

E first = q.peek();

if (first == null) //没有元素,让出线程,等待java.lang.Thread.State#WAITING

available.await();

else {

long delay = first.getDelay(NANOSECONDS);

if (delay <= 0) // 已到期,元素出队

return q.poll();

first = null; // don't retain ref while waiting

if (leader != null)

available.await();// 其它线程在leader线程TIMED_WAITING期间,会进入等待状态,这样可以只有一个线程去等待到时唤醒,避免大量唤醒操作

else {

Thread thisThread = Thread.currentThread();

leader = thisThread;

try {

available.awaitNanos(delay);// 等待剩余时间后,再尝试获取元素,他在等待期间,由于leader是当前线程,所以其它线程会等待。

} finally {

if (leader == thisThread) leader = null;

} } } } }

finally {

if (leader == null && q.peek() != null) available.signal();

lock.unlock(); } }

示例:

package org.dromara.hmily.demo.springcloud.account.service;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

import java.util.concurrent.DelayQueue;

import java.util.concurrent.Delayed;

import java.util.concurrent.TimeUnit;

/**

* @description: 延时队列测试

* @author: hh

*/

public class DelayedQueneTest {

public static void main(String[] args) throws InterruptedException {

Item item1 = new Item("item1", 5, TimeUnit.SECONDS);

Item item2 = new Item("item2",10, TimeUnit.SECONDS);

Item item3 = new Item("item3",15, TimeUnit.SECONDS);

DelayQueue<Item> queue = new DelayQueue<>();

queue.put(item1);

queue.put(item2);

queue.put(item3);

System.out.println("begin time:" + LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));

for (int i = 0; i < 3; i++) {

Item take = queue.take();

System.out.format("name:{%s}, time:{%s}\n",take.name, LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));

}

}

}

class Item implements Delayed{

/* 触发时间*/

private long time;

String name;

public Item(String name, long time, TimeUnit unit) {

this.name = name;

this.time = System.currentTimeMillis() + (time > 0? unit.toMillis(time): 0);

}

@Override

public long getDelay(TimeUnit unit) {

return time - System.currentTimeMillis();

}

@Override

public int compareTo(Delayed o) {

Item item = (Item) o;

long diff = this.time - item.time;

if (diff <= 0) {// 改成>=会造成问题

return -1;

}else {

return 1;

}

}

@Override

public String toString() {

return "Item{" +

"time=" + time +

", name='" + name + '\'' +

'}';

}

}

 运行结果:每5秒取出一个

begin time:2019-05-31T11:58:24.445

name:{item1}, time:{2019-05-31T11:58:29.262}

name:{item2}, time:{2019-05-31T11:58:34.262}

name:{item3}, time:{2019-05-31T11:58:39.262}

修改compareTo方法 diff >= 0 后的运行结果: 在15秒之后几乎同时取出,

begin time:2019-05-31T12:02:50.157

name:{item3}, time:{2019-05-31T12:03:04.959}

name:{item2}, time:{2019-05-31T12:03:04.999}

name:{item1}, time:{2019-05-31T12:03:05}

标签: #java使用队列