java實作訊息佇列以及延遲訊息（佇列DelayQueue）

1.java實作延遲訊息（佇列DelayQueue）

DelayQueue是一個支持延時獲取元素的無界阻塞佇列，佇列使用PriorityQueue來實作，佇列中的元素必須實作Delayed介面，在創建元素時可以指定多久才能從佇列中獲取當前元素，只有在延遲期滿時才能從佇列中提取元素，

快取系統的設計：這里使用DelayQueue保存快取元素的有效期，一個執行緒（生產者）設定失效實作回圈添加訊息，使用一個執行緒（消費者）回圈查詢
DelayQueue，一旦能從DelayQueue中獲取元素時，表示快取有效期到了

應用場景：

• 訊息生產和消費有時間視窗要求，例如在電商交易中超時未支付關閉訂單的場景，在訂單創建時會發送一條延時訊息，這條訊息將會在 30 分鐘以后投遞給消費者，消費者收到此訊息后需要判斷對應的訂單是否已完成支付，如支付未完成，則關閉訂單，如已完成支付則忽略，
• 通過訊息觸發一些定時任務，例如在某一固定時間點向用戶發送提醒訊息，

前提條件：放置在DelayQueue的元素需要實作Delayed介面，Delayed介面使物件成為延遲物件，它使存放在DelayQueue類中的物件具有了激活日期

CompareTo(Delayed o)：Delayed介面繼承了Comparable介面，因此有了這個方法，
getDelay(TimeUnit unit):這個方法回傳到激活日期的剩余時間，時間單位由單位引數指定，

2.實作Delayed介面

package com.violet.Queue;

import java.time.format.DateTimeFormatter;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class MessageData implements Delayed {
    private static final AtomicLong atomic = new AtomicLong(0);
    // 資料的失效時間點
    private final long time;
    // 序號
    private final long seqno;

    /**
     * @param deadline 資料失效時間點
     */
    public MessageData(long deadline) {
        this.time = deadline;
        //序號自增類似于i++ ，這里使用AtomicLong實作原子操作
        this.seqno = atomic.getAndIncrement();
    }
    /**
     * 回傳剩余有效時間
     *
     * @param unit 時間單位
     */
    @Override
    public long getDelay(TimeUnit unit) {
    	//查看是不是當這個時間到期時，訊息被消費
        System.out.println(unit.convert(this.time - System.currentTimeMillis(), TimeUnit.NANOSECONDS));
        return unit.convert(this.time - System.currentTimeMillis(), TimeUnit.NANOSECONDS);
    }

    /**
     * 比較兩個Delayed物件的大小, 比較順序如下:
     * 1. 如果是物件本身, 回傳0;
     * 2. 比較失效時間點, 先失效的回傳-1,后失效的回傳1;
     * 3. 比較元素序號, 序號小的回傳-1, 否則回傳1.
     * 4. 非Data型別元素, 比較剩余有效時間, 剩余有效時間小的回傳-1,大的回傳1,相同回傳0
     */
    @Override
    public int compareTo(Delayed other) {
        if (other == this)  // compare zero if same object
            return 0;
        if (other instanceof MessageData) {
            MessageData x = (MessageData) other;
            // 優先比較失效時間
            long diff = this.time - x.time;
            if (diff < 0)
                return -1;
            else if (diff > 0)
                return 1;
            else if (this.seqno < x.seqno)    // 剩余時間相同則比較序號
                return -1;
            else
                return 1;
        }
        // 一般不會執行到此處，除非元素不是MessageData型別
        long diff = this.getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS);
        return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
    }

    @Override
    public String toString() {
        return "Data{" +
            "time=" + time +
            ", seqno=" + seqno +
            "}, isValid=" + isValid();
    }

    private boolean isValid() {
        return this.getDelay(TimeUnit.NANOSECONDS) > 0;
    }

}

3.生產者

package com.violet.Queue;

import java.util.Date;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadLocalRandom;

public class Producer implements Runnable {
    private final DelayQueue<MessageData> queue;

    public Producer(DelayQueue<MessageData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            //選取系統當前時間+隨機生成的時間 來設定訊息失效時間
            long currentTime = System.currentTimeMillis();
            long validTime = ThreadLocalRandom.current().nextLong(1000L, 7000L);
            MessageData data = new MessageData(currentTime + validTime);
            queue.put(data);
            System.out.println(Thread.currentThread().getName() + ": put " + data);
            try {
              //為了效果顯著這里將執行緒停的時間長一點
                Thread.sleep(1000000000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

4.消費者

package com.violet.Queue;

import java.util.concurrent.DelayQueue;

public class Consumer implements Runnable {
    private final DelayQueue<MessageData> queue;
    public Consumer(DelayQueue<MessageData> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            try {
                MessageData data = queue.take();
                System.out.println(Thread.currentThread().getName() + ": take " + data);
                Thread.yield();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

5.main方法

package com.violet.Queue;

import java.util.concurrent.DelayQueue;

public class Main {
    public static void main(String[] args) {
        DelayQueue<MessageData> queue = new DelayQueue<>();
        Thread c1 = new Thread(new Consumer(queue), "consumer");
        Thread p1 = new Thread(new Producer(queue), "producer");
        c1.start();
        p1.start();
    }
}

6.效果