java中DelayQueue的使用

時間 2020-05-12

原文原文鏈接

java中DelayQueue的使用java

簡介

今天給你們介紹一下DelayQueue,DelayQueue是BlockingQueue的一種，因此它是線程安全的，DelayQueue的特色就是插入Queue中的數據能夠按照自定義的delay時間進行排序。只有delay時間小於0的元素纔可以被取出。git

DelayQueue

先看一下DelayQueue的定義：github

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

從定義能夠看到，DelayQueue中存入的對象都必須是Delayed的子類。安全

Delayed繼承自Comparable，而且須要實現一個getDelay的方法。dom

爲何這樣設計呢？ide

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

DelayQueue的應用

DelayQueue通常用於生產者消費者模式，咱們下面舉一個具體的例子。this

首先要使用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，那麼在線程池關閉以前插入的元素是不會過時的，也就是說消費者是沒法獲取到結果的。