Java延迟队列

DelayQueue 实现 BlockingQueue 接口。 DelayQueue 中的元素必须保留一定的时间。

DelayQueue 使用一个名为 Delayed 的接口来获取延迟时间。

该接口在java.util.concurrent包中。 其声明如下：

public interface  Delayed  extends Comparable  {
   long  getDelay(TimeUnit timeUnit);
}

它扩展了 Comparable 接口，它的 compareTo()方法接受一个Delayed对象。

DelayQueue 调用每个元素的 getDelay()方法来获取元素必须保留多长时间。 DelayQueue 将传递 TimeUnit 到此方法。

当 getDelay()方法返回一个零或一个负数时，是元素离开队列的时间。

队列通过调用元素的 compareTo()方法确定要弹出的那个。 此方法确定要从队列中删除的过期元素的优先级。

以下代码显示了如何使用DelayQueue。

import static java.time.temporal.ChronoUnit.MILLIS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

import java.time.Instant;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

class DelayedJob implements Delayed {
  private Instant scheduledTime;
  String jobName;

  public DelayedJob(String jobName, Instant scheduledTime) {
    this.scheduledTime = scheduledTime;
    this.jobName = jobName;
  }

  @Override
  public long getDelay(TimeUnit unit) {
    long delay = MILLIS.between(Instant.now(), scheduledTime);
    long returnValue = unit.convert(delay, MILLISECONDS);
    return returnValue;
  }

  @Override
  public int compareTo(Delayed job) {
    long currentJobDelay = this.getDelay(MILLISECONDS);
    long jobDelay = job.getDelay(MILLISECONDS);

    int diff = 0;
    if (currentJobDelay > jobDelay) {
      diff = 1;
    } else if (currentJobDelay < jobDelay) {
      diff = -1;
    }
    return diff;
  }

  @Override
  public String toString() {
    String str = this.jobName + ", " + "Scheduled Time:  "
        + this.scheduledTime;
    return str;
  }
}
public class Main {
  public static void main(String[] args) throws InterruptedException {
    BlockingQueue queue = new DelayQueue<>();
    Instant now = Instant.now();

    queue.put(new DelayedJob("A", now.plusSeconds(9)));
    queue.put(new DelayedJob("B", now.plusSeconds(3)));
    queue.put(new DelayedJob("C", now.plusSeconds(6)));
    queue.put(new DelayedJob("D", now.plusSeconds(1)));

    while (queue.size() > 0) {
      System.out.println("started...");
      DelayedJob job = queue.take();
      System.out.println("Job: " + job);
    }
    System.out.println("Finished.");
  }
}

上面的代码生成以下结果。