簡介

今天給大家介紹一下DelayQueue,DelayQueue是BlockingQueue的一種，所以它是執行緒安全的，DelayQueue的特點就是插入Queue中的資料可以按照自定義的delay時間進行排序。只有delay時間小於0的元素才能夠被取出。

DelayQueue

先看一下DelayQueue的定義：

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
  implements BlockingQueue<E>

從定義可以看到，DelayQueue中存入的物件都必須是Delayed的子類。

Delayed繼承自Comparable，並且需要實現一個getDelay的方法。

為什麼這樣設計呢？

因為DelayQueue的底層儲存是一個PriorityQueue，在之前的文章中我們講過了，PriorityQueue是一個可排序的Queue，其中的元素必須實現Comparable方法。而getDelay方法則用來判斷排序後的元素是否可以從Queue中取出。

DelayQueue的應用

DelayQueue一般用於生產者消費者模式，我們下面舉一個具體的例子。

首先要使用DelayQueue，必須自定義一個Delayed物件：

@Data
public class DelayedUser implements Delayed {
  private String name;
  private long avaibleTime;

  public DelayedUser(String name,long delayTime){
    this.name=name;
    //avaibleTime = 當前時間+ delayTime
    this.avaibleTime=delayTime + System.currentTimeMillis();

  }

  @Override
  public long getDelay(TimeUnit unit) {
    //判斷avaibleTime是否大於當前系統時間，並將結果轉換成MILLISECONDS
    long diffTime= avaibleTime- System.currentTimeMillis();
    return unit.convert(diffTime,TimeUnit.MILLISECONDS);
  }

  @Override
  public int compareTo(Delayed o) {
    //compareTo用在DelayedUser的排序
    return (int)(this.avaibleTime - ((DelayedUser) o).getAvaibleTime());
  }
}

上面的物件中，我們需要實現getDelay和compareTo方法。

接下來我們建立一個生產者：

@Slf4j
@Data
@AllArgsConstructor
class DelayedQueueProducer implements Runnable {
  private DelayQueue<DelayedUser> delayQueue;

  private Integer messageCount;

  private long delayedTime;

  @Override
  public void run() {
    for (int i = 0; i < messageCount; i++) {
      try {
        DelayedUser delayedUser = new DelayedUser(
            new Random().nextInt(1000)+"",delayedTime);
        log.info("put delayedUser {}",delayedUser);
        delayQueue.put(delayedUser);
        Thread.sleep(500);
      } catch (InterruptedException e) {
        log.error(e.getMessage(),e);
      }
    }
  }
}

在生產者中，我們每隔0.5秒建立一個新的DelayedUser物件，併入Queue。

再建立一個消費者：

@Slf4j
@Data
@AllArgsConstructor
public class DelayedQueueConsumer implements Runnable {

  private DelayQueue<DelayedUser> delayQueue;

  private int messageCount;

  @Override
  public void run() {
    for (int i = 0; i < messageCount; i++) {
      try {
        DelayedUser element = delayQueue.take();
        log.info("take {}",element );
      } catch (InterruptedException e) {
        log.error(e.getMessage(),e);
      }
    }
  }
}

在消費者中，我們迴圈從queue中獲取物件。

最後看一個呼叫的例子：

  @Test
  public void useDelayedQueue() throws InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(2);

    DelayQueue<DelayedUser> queue = new DelayQueue<>();
    int messageCount = 2;
    long delayTime = 500;
    DelayedQueueConsumer consumer = new DelayedQueueConsumer(
        queue,messageCount);
    DelayedQueueProducer producer = new DelayedQueueProducer(
        queue,messageCount,delayTime);

    // when
    executor.submit(producer);
    executor.submit(consumer);

    // then
    executor.awaitTermination(5,TimeUnit.SECONDS);
    executor.shutdown();

  }

上面的測試例子中，我們定義了兩個執行緒的執行緒池，生產者產生兩條訊息，delayTime設定為0.5秒，也就是說0.5秒之後，插入的物件能夠被獲取到。

執行緒池在5秒之後會被關閉。

執行看下結果：

[pool-1-thread-1] INFO com.flydean.DelayedQueueProducer - put delayedUser DelayedUser(name=917,avaibleTime=1587623188389)
[pool-1-thread-2] INFO com.flydean.DelayedQueueConsumer - take DelayedUser(name=917,avaibleTime=1587623188389)
[pool-1-thread-1] INFO com.flydean.DelayedQueueProducer - put delayedUser DelayedUser(name=487,avaibleTime=1587623188899)
[pool-1-thread-2] INFO com.flydean.DelayedQueueConsumer - take DelayedUser(name=487,avaibleTime=1587623188899)

我們看到訊息的put和take是交替進行的，符合我們的預期。

如果我們做下修改，將delayTime修改為50000，那麼線上程池關閉之前插入的元素是不會過期的，也就是說消費者是無法獲取到結果的。

總結

DelayQueue是一種有奇怪特性的BlockingQueue，可以在需要的時候使用。

本文的例子https://github.com/ddean2009/learn-java-collections

以上就是詳解java中DelayQueue的使用的詳細內容，更多關於java DelayQueue的資料請關注我們其它相關文章！