Java中延遲隊列的原理是什么

這篇文章將為大家詳細講解有關Java中延遲隊列的原理是什么，文章內容質量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關知識有一定的了解。

Java是什么

Java是一門面向對象編程語言，可以編寫桌面應用程序、Web應用程序、分布式系統和嵌入式系統應用程序。

一、消息體

package com.delqueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class Message implements Delayed {
    private int id;
    private String body; // 消息內容
    private long excuteTime;// 延遲時長，這個是必須的屬性因為要按照這個判斷延時時長。
    public int getId() {
        return id;
    }
    public String getBody() {
        return body;
    }
    public long getExcuteTime() {
        return excuteTime;
    }
    public Message(int id, String body, long delayTime) {
        this.id = id;
        this.body = body;
        this.excuteTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();
    }
    // 自定義實現比較方法返回 1 0 -1三個參數
    @Override
    public int compareTo(Delayed delayed) {
        Message msg = (Message) delayed;
        return Integer.valueOf(this.id) > Integer.valueOf(msg.id) ? 1
                : (Integer.valueOf(this.id) < Integer.valueOf(msg.id) ? -1 : 0);
    }
    // 延遲任務是否到時就是按照這個方法判斷如果返回的是負數則說明到期否則還沒到期
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.excuteTime - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
}

二、消息消費者

package com.delqueue;
import java.util.concurrent.DelayQueue;
public class Consumer implements Runnable {
    // 延時隊列 ,消費者從其中獲取消息進行消費
    private DelayQueue<Message> queue;
    public Consumer(DelayQueue<Message> queue) {
        this.queue = queue;
    }
    @Override
    public void run() {
        while (true) {
            try {
                Message take = queue.take();
                System.out.println("消費消息id：" + take.getId() + " 消息體：" + take.getBody());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

三、延時隊列

package com.delqueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class DelayQueueTest {
     public static void main(String[] args) {
        // 創建延時隊列
        DelayQueue<Message> queue = new DelayQueue<Message>();
        // 添加延時消息,m1 延時3s
        Message m1 = new Message(1, "world", 3000);
        // 添加延時消息,m2 延時10s
        Message m2 = new Message(2, "hello", 10000);
        //將延時消息放到延時隊列中
        queue.offer(m2);
        queue.offer(m1);
        // 啟動消費線程 消費添加到延時隊列中的消息，前提是任務到了延期時間
        ExecutorService exec = Executors.newFixedThreadPool(1);
        exec.execute(new Consumer(queue));
        exec.shutdown();
      }
}

將消息體放入延遲隊列中，在啟動消費者線程去消費延遲隊列中的消息，如果延遲隊列中的消息到了延遲時間則可以從中取出消息否則無法取出消息也就無法消費。

這就是延遲隊列demo，下面我們來說說在真實環境下的使用。

使用場景描述：

在打車軟件中對訂單進行派單的流程，當有訂單的時候給該訂單篩選司機，然后給當訂單綁定司機，但是有時運氣沒那么好，訂單進來后第一次沒有篩選到合適的司機，但我們也不能就此結束派單，而是將該訂單的信息放到延時隊列中過個2秒鐘在進行一次，其實這個2秒鐘就是一個延遲，所以這里我們就可以使用延時隊列來實現……

下面看看簡單的流程圖：

下面來看看具體代碼實現：

在項目中有如下幾個類：第一 、任務類   第二、按照任務類組裝的消息體類  第三、延遲隊列管理類

任務類即執行篩選司機、綁單、push消息的任務類

package com.test.delayqueue;
/**
 * 具體執行相關業務的業務類
 * @author whd
 * @date 2017年9月25日 上午12:49:32
 */
public class DelayOrderWorker implements Runnable {
    @Override
    public void run() {
        // TODO Auto-generated method stub
        //相關業務邏輯處理
        System.out.println(Thread.currentThread().getName()+" do something ……");
    }
}

消息體類，在延時隊列中這個實現了Delayed接口的消息類是比不可少的，實現接口時有一個getDelay(TimeUnit unit)方法，這個方法就是判斷是否到期的

這里定義的是一個泛型類，所以可以將我們上面的任務類作為其中的task，這樣就將任務類分裝成了一個消息體

package com.test.delayqueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/**
 * 延時隊列中的消息體將任務封裝為消息體
 *
 * @author whd
 * @date 2017年9月25日 上午12:48:30
 * @param <T>
 */
public class DelayOrderTask<T extends Runnable> implements Delayed {
    private final long time;
    private final T task; // 任務類，也就是之前定義的任務類
    /**
     * @param timeout
     *      超時時間(秒)
     * @param task
     *      任務
     */
    public DelayOrderTask(long timeout, T task) {
        this.time = System.nanoTime() + timeout;
        this.task = task;
    }
    @Override
    public int compareTo(Delayed o) {
        // TODO Auto-generated method stub
        DelayOrderTask other = (DelayOrderTask) o;
        long diff = time - other.time;
        if (diff > 0) {
            return 1;
        } else if (diff < 0) {
            return -1;
        } else {
            return 0;
        }
    }
    @Override
    public long getDelay(TimeUnit unit) {
        // TODO Auto-generated method stub
        return unit.convert(this.time - System.nanoTime(), TimeUnit.NANOSECONDS);
    }
    @Override
    public int hashCode() {
        return task.hashCode();
    }
    public T getTask() {
        return task;
    }
}

延時隊列管理類，這個類主要就是將任務類封裝成消息并并添加到延時隊列中，以及輪詢延時隊列從中取出到時的消息體，在獲取任務類放到線程池中執行任務

package com.test.delayqueue;
import java.util.Map;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
 * 延時隊列管理類，用來添加任務、執行任務
 *
 * @author whd
 * @date 2017年9月25日 上午12:44:59
 */
public class DelayOrderQueueManager {
    private final static int DEFAULT_THREAD_NUM = 5;
    private static int thread_num = DEFAULT_THREAD_NUM;
    // 固定大小線程池
    private ExecutorService executor;
    // 守護線程
    private Thread daemonThread;
    // 延時隊列
    private DelayQueue<DelayOrderTask<?>> delayQueue;
    private static final AtomicLong atomic = new AtomicLong(0);
    private final long n = 1;
    private static DelayOrderQueueManager instance = new DelayOrderQueueManager();
    private DelayOrderQueueManager() {
        executor = Executors.newFixedThreadPool(thread_num);
        delayQueue = new DelayQueue<>();
        init();
    }
    public static DelayOrderQueueManager getInstance() {
        return instance;
    }
    /**
     * 初始化
     */
    public void init() {
        daemonThread = new Thread(() -> {
            execute();
        });
        daemonThread.setName("DelayQueueMonitor");
        daemonThread.start();
    }
    private void execute() {
        while (true) {
            Map<Thread, StackTraceElement[]> map = Thread.getAllStackTraces();
            System.out.println("當前存活線程數量:" + map.size());
            int taskNum = delayQueue.size();
            System.out.println("當前延時任務數量:" + taskNum);
            try {
                // 從延時隊列中獲取任務
                DelayOrderTask<?> delayOrderTask = delayQueue.take();
                if (delayOrderTask != null) {
                    Runnable task = delayOrderTask.getTask();
                    if (null == task) {
                        continue;
                    }
                    // 提交到線程池執行task
                    executor.execute(task);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    /**
     * 添加任務
     *
     * @param task
     * @param time
     *      延時時間
     * @param unit
     *      時間單位
     */
    public void put(Runnable task, long time, TimeUnit unit) {
        // 獲取延時時間
        long timeout = TimeUnit.NANOSECONDS.convert(time, unit);
        // 將任務封裝成實現Delayed接口的消息體
        DelayOrderTask<?> delayOrder = new DelayOrderTask<>(timeout, task);
        // 將消息體放到延時隊列中
        delayQueue.put(delayOrder);
    }
    /**
     * 刪除任務
     *
     * @param task
     * @return
     */
    public boolean removeTask(DelayOrderTask task) {
        return delayQueue.remove(task);
    }
}

測試類

package com.delqueue;
import java.util.concurrent.TimeUnit;
import com.test.delayqueue.DelayOrderQueueManager;
import com.test.delayqueue.DelayOrderWorker;
public class Test {
    public static void main(String[] args) {
        DelayOrderWorker work1 = new DelayOrderWorker();// 任務1
        DelayOrderWorker work2 = new DelayOrderWorker();// 任務2
        DelayOrderWorker work3 = new DelayOrderWorker();// 任務3
        // 延遲隊列管理類，將任務轉化消息體并將消息體放入延遲對列中等待執行
        DelayOrderQueueManager manager = DelayOrderQueueManager.getInstance();
        manager.put(work1, 3000, TimeUnit.MILLISECONDS);
        manager.put(work2, 6000, TimeUnit.MILLISECONDS);
        manager.put(work3, 9000, TimeUnit.MILLISECONDS);
    }
}

關于Java中延遲隊列的原理是什么就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。