簡介

今天給大家介紹一下DelayQueue,DelayQueue是BlockingQueue的一種，所以它是線程安全的，DelayQueue的特點就是插入Queue中的數(shù)據(jù)可以按照自定義的delay時間進行排序。只有delay時間小于0的元素才能夠被取出。

DelayQueue

先看一下DelayQueue的定義：

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

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

Delayed繼承自Comparable，并且需要實現(xiàn)一個getDelay的方法。

為什么這樣設(shè)計呢？

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

DelayQueue的應用

DelayQueue一般用于生產(chǎn)者消費者模式，我們下面舉一個具體的例子。

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

@Datapublic 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是否大于當前系統(tǒng)時間，并將結(jié)果轉(zhuǎn)換成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()); }}

上面的對象中，我們需要實現(xiàn)getDelay和compareTo方法。

接下來我們創(chuàng)建一個生產(chǎn)者：

@Slf4j@Data@AllArgsConstructorclass 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); } } }}

在生產(chǎn)者中，我們每隔0.5秒創(chuàng)建一個新的DelayedUser對象，并入Queue。

再創(chuàng)建一個消費者：

@Slf4j@Data@AllArgsConstructorpublic 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); } } }}

在消費者中，我們循環(huán)從queue中獲取對象。

最后看一個調(diào)用的例子：

@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(); }

上面的測試例子中，我們定義了兩個線程的線程池，生產(chǎn)者產(chǎn)生兩條消息，delayTime設(shè)置為0.5秒，也就是說0.5秒之后，插入的對象能夠被獲取到。

線程池在5秒之后會被關(guān)閉。

運行看下結(jié)果：

[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，那么在線程池關(guān)閉之前插入的元素是不會過期的，也就是說消費者是無法獲取到結(jié)果的。

總結(jié)

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

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

以上就是詳解java中DelayQueue的使用的詳細內(nèi)容，更多關(guān)于java DelayQueue的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！