最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Java延遲隊列原理與用法實例詳解

 更新時間:2018年09月03日 11:45:35   作者:QH_JAVA  
這篇文章主要介紹了Java延遲隊列原理與用法,結合實例形式詳細分析了延遲隊列的概念、原理、功能及具體使用方法,需要的朋友可以參考下

本文實例講述了Java延遲隊列原理與用法。分享給大家供大家參考,具體如下:

延時隊列,第一他是個隊列,所以具有對列功能第二就是延時,這就是延時對列,功能也就是將任務放在該延時對列中,只有到了延時時刻才能從該延時對列中獲取任務否則獲取不到……

應用場景比較多,比如延時1分鐘發(fā)短信,延時1分鐘再次執(zhí)行等,下面先看看延時隊列demo之后再看延時隊列在項目中的使用:

簡單的延時隊列要有三部分:第一實現了Delayed接口的消息體、第二消費消息的消費者、第三存放消息的延時隊列,那下面就來看看延時隊列demo。

一、消息體

package com.delqueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/**
 * 消息體定義 實現Delayed接口就是實現兩個方法即compareTo 和 getDelay最重要的就是getDelay方法,這個方法用來判斷是否到期……
 *
 * @author whd
 * @date 2017年9月24日 下午8:57:14
 */
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) {
        // 創(chuàng)建延時隊列
        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,下面我們來說說在真實環(huán)境下的使用。

使用場景描述:

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

下面看看簡單的流程圖:

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

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

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

package com.test.delayqueue;
/**
 * 具體執(zhí)行相關業(yè)務的業(yè)務類
 * @author whd
 * @date 2017年9月25日 上午12:49:32
 */
public class DelayOrderWorker implements Runnable {
    @Override
    public void run() {
        // TODO Auto-generated method stub
        //相關業(yè)務邏輯處理
        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;
    }
}

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

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;
/**
 * 延時隊列管理類,用來添加任務、執(zhí)行任務
 *
 * @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;
                    }
                    // 提交到線程池執(zhí)行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
        // 延遲隊列管理類,將任務轉化消息體并將消息體放入延遲對列中等待執(zhí)行
        DelayOrderQueueManager manager = DelayOrderQueueManager.getInstance();
        manager.put(work1, 3000, TimeUnit.MILLISECONDS);
        manager.put(work2, 6000, TimeUnit.MILLISECONDS);
        manager.put(work3, 9000, TimeUnit.MILLISECONDS);
    }
}

OK 這就是項目中的具體使用情況,當然具體內容被忽略,整體框架就是這樣,還有這里使用java的延時隊列但是這種方式是有問題的如果如果down機則會出現任務丟失,所以也可以考慮使用mq、redis來實現……

更多關于java算法相關內容感興趣的讀者可查看本站專題:《Java數據結構與算法教程》、《Java操作DOM節(jié)點技巧總結》、《Java文件與目錄操作技巧匯總》和《Java緩存操作技巧匯總

希望本文所述對大家java程序設計有所幫助。

相關文章

最新評論

依兰县| 通化县| 黎城县| 贵州省| 志丹县| 兴文县| 安溪县| 灌阳县| 万山特区| 仪陇县| 黄浦区| 贵定县| 高安市| 台中县| 永丰县| 噶尔县| 桃园市| 高平市| 沙河市| 宿松县| 隆回县| 阿坝| 乌鲁木齐县| 梁山县| 白沙| 马边| 临桂县| 溆浦县| 潮安县| 达州市| 临沭县| 枣阳市| 云和县| 普安县| 中阳县| 陆丰市| 罗田县| 洛浦县| 永吉县| 宁都县| 广东省|