本文實例講述了java延遲隊列原理與用法。分享給大家供大家參考，具體如下：
延時隊列，第一他是個隊列，所以具有對列功能第二就是延時，這就是延時對列，功能也就是將任務放在該延時對列中，只有到了延時時刻才能從該延時對列中獲取任務否則獲取不到……
應用場景比較多，比如延時1分鐘發短信，延時1分鐘再次執行等，下面先看看延時隊列demo之后再看延時隊列在項目中的使用：
簡單的延時隊列要有三部分：第一實現了delayed接口的消息體、第二消費消息的消費者、第三存放消息的延時隊列，那下面就來看看延時隊列demo。
一、消息體
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
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); }} |
二、消息消費者
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
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(); } } }} |
三、延時隊列
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
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消息的任務類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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，這樣就將任務類分裝成了一個消息體
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
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; }} |
延時隊列管理類，這個類主要就是將任務類封裝成消息并并添加到延時隊列中，以及輪詢延時隊列從中取出到時的消息體，在獲取任務類放到線程池中執行任務
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
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); }} |
測試類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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); }} |
ok 這就是項目中的具體使用情況，當然具體內容被忽略，整體框架就是這樣，還有這里使用java的延時隊列但是這種方式是有問題的如果如果down機則會出現任務丟失，所以也可以考慮使用mq、redis來實現……
希望本文所述對大家java程序設計有所幫助。
原文鏈接：https://blog.csdn.net/QH_JAVA/article/details/78080969
