實(shí)現(xiàn)延時(shí)任務(wù)有很多的方法，網(wǎng)上關(guān)于延時(shí)任務(wù)的實(shí)現(xiàn)的文章已經(jīng)不少了。比如：實(shí)現(xiàn)延時(shí)任務(wù)的10種方法等等。但是這些文章基本上都是將方法大概的列舉一下，給出部分示例代碼，對(duì)于有經(jīng)驗(yàn)的老程序員可能一看就知道該怎么去把它實(shí)現(xiàn)完整，但是對(duì)于初學(xué)者來(lái)說(shuō)不夠友好。

小概念：什么是延時(shí)任務(wù)？舉個(gè)例子：你買(mǎi)了一張火車(chē)票，必須在30分鐘之內(nèi)付款，否則該訂單被自動(dòng)取消?！赣唵?0分鐘不付款自動(dòng)取消，這個(gè)任務(wù)就是一個(gè)延時(shí)任務(wù)?！?/p>

一、DelayQueue的應(yīng)用原理

DelayQueue是一個(gè)無(wú)界的BlockingQueue的實(shí)現(xiàn)類，用于放置實(shí)現(xiàn)了Delayed接口的對(duì)象，其中的對(duì)象只能在其到期時(shí)才能從隊(duì)列中取走。

• BlockingQueue即阻塞隊(duì)列，java提供的面向多線程安全的隊(duì)列數(shù)據(jù)結(jié)構(gòu)，當(dāng)隊(duì)列內(nèi)元素?cái)?shù)量為0的時(shí)候，試圖從隊(duì)列內(nèi)獲取元素的線程將被阻塞或者拋出異常。
• 這里的“無(wú)界”隊(duì)列，是指隊(duì)列的元素?cái)?shù)量不存在上限，隊(duì)列的容量會(huì)隨著元素?cái)?shù)量的增加而擴(kuò)容。

DelayQueue實(shí)現(xiàn)了BlockingQueue接口，所以具有無(wú)界、阻塞的特點(diǎn)，除此之外它自己的核心特點(diǎn)就是：

• 「放入該隊(duì)列的延時(shí)任務(wù)對(duì)象，只要到達(dá)延時(shí)時(shí)間之后才能被取到」。
• DelayQueue 不接收null元素
• 「DelayQueue 只接受那些實(shí)現(xiàn)了java.util.concurrent.Delayed接口的對(duì)象」

二、訂單延時(shí)任務(wù)的實(shí)現(xiàn)

了解了DelayQueue的特點(diǎn)之后，我們就可以利用它來(lái)實(shí)現(xiàn)延時(shí)任務(wù)了，實(shí)現(xiàn)java.util.concurrent.Delayed接口。

import?org.jetbrains.annotations.NotNull;
import?java.text.SimpleDateFormat;
import?java.util.Date;
import?java.util.concurrent.Delayed;
import?java.util.concurrent.TimeUnit;
/**
?*?延時(shí)訂單任務(wù)
?*/
public?class?OrderDelayObject?implements?Delayed?{
??private?String?name;
??private?long?delayTime;???//延時(shí)時(shí)間
??//實(shí)際業(yè)務(wù)中這里傳訂單信息對(duì)象，我這里只做demo，所以使用字符串了
??private?String?order;
??public?OrderDelayObject(String?name,?long?delayTime,?String?order)?{
????this.name?=?name;
????//延時(shí)時(shí)間加上當(dāng)前時(shí)間
????this.delayTime?=?System.currentTimeMillis()?+?delayTime;
????this.order?=?order;
??}
??//獲取延時(shí)任務(wù)的倒計(jì)時(shí)時(shí)間
??@Override
??public?long?getDelay(TimeUnit?unit)?{
????long?diff?=?delayTime?-?System.currentTimeMillis();
????return?unit.convert(diff,?TimeUnit.MILLISECONDS);
??}
??//延時(shí)任務(wù)隊(duì)列，按照延時(shí)時(shí)間元素排序，實(shí)現(xiàn)Comparable接口
??@Override
??public?int?compareTo(@NotNull?Delayed?obj)?{
????return?Long.compare(this.delayTime,?((OrderDelayObject)?obj).delayTime);
??}
??@Override
??public?String?toString()?{
????Date?date?=?new?Date(delayTime);
????SimpleDateFormat?sd?=?new?SimpleDateFormat("yyyy-MM-dd?HH:mm:ss");
????return?"\nOrderDelayObject:{"
????????????+?"name="?+?name
????????????+?",?time="?+?sd.format(date)
????????????+?",?order="?+?order
????????????+?"}";
??}
}?

上文類中的order為訂單信息對(duì)象，在實(shí)際的業(yè)務(wù)開(kāi)發(fā)過(guò)程中應(yīng)該是傳遞訂單信息，用于取消訂單業(yè)務(wù)的實(shí)現(xiàn)（訂單30分鐘不付款自動(dòng)取消）。

Delayed接口繼承自 Comparable接口，所以需要實(shí)現(xiàn)compareTo方法，用于延時(shí)任務(wù)在隊(duì)列中按照“延時(shí)時(shí)間”進(jìn)行排序。

getDelay方法是Delayed接口方法，實(shí)現(xiàn)該方法提供獲取延時(shí)任務(wù)的倒計(jì)時(shí)時(shí)間

三、訂單處理

首先我們需要一個(gè)容器，永久保存延時(shí)任務(wù)隊(duì)列，如果是Spring開(kāi)發(fā)環(huán)境我們可以這樣做。

@Bean("orderDelayQueue")
public DelayQueue<OrderDelayObject> orderDelayQueue(){
    return new DelayQueue<OrderDelayObject>();
}

當(dāng)用戶下單的時(shí)候，將訂單下單任務(wù)放入延時(shí)隊(duì)列

@Resource
private?DelayQueue<OrderDelayObject>?orderDelayQueue;
//發(fā)起訂單下單的時(shí)候?qū)⒂唵窝菔緦?duì)象放入orderDelayQueue
orderDelayQueue.add(
????????new?OrderDelayObject(
????????????????"訂單延時(shí)取消任務(wù)",
????????????????30?*?60?*?1000,????//延時(shí)30分鐘
????????????????"延時(shí)任務(wù)訂單對(duì)象信息"
????????)
);

系統(tǒng)內(nèi)開(kāi)啟一個(gè)線程，不斷的從隊(duì)列中獲取消息，獲取到之后對(duì)延時(shí)消息進(jìn)行處理。DelayQueue的take方法從隊(duì)列中獲取延時(shí)任務(wù)對(duì)象，如果隊(duì)列元素?cái)?shù)量為0，或者沒(méi)有到達(dá)“延時(shí)時(shí)間的任務(wù)”，該線程會(huì)被阻塞。

@Component
public class DelayObjectConsumer<OrderDelayObject> implements InitializingBean {
? @Resource
? private DelayQueue<OrderDelayObject> orderDelayQueue;
? @Override
? public void afterPropertiesSet() throws Exception {
? ? new Thread(() -> {
? ? ? while (true) {
? ? ? ? OrderDelayObject task = orderDelayQueue.take();
? ? ? ? System.out.println(task.toString());
? ? ? ? System.out.println(task.getOrder());
? ? ? ? //根據(jù)order訂單信息，去查詢?cè)撚唵蔚闹Ц缎畔?
? ? ? ? //如果用戶沒(méi)有進(jìn)行支付，將訂單從數(shù)據(jù)庫(kù)中關(guān)閉
? ? ? ? //如果訂單并發(fā)量比較大，這里可以考慮異步或線程池的方式進(jìn)行處理
? ? ? }
? ? }).start();
? }
}

需要說(shuō)明的是，這里的while-true循環(huán)的延時(shí)任務(wù)處理是順序執(zhí)行的，在訂單并發(fā)量比較大的時(shí)候，需要考慮異步處理的方式完成訂單的關(guān)閉操作。我之前寫(xiě)過(guò)一個(gè)SpringBoot的可觀測(cè)、易配置的線程池開(kāi)源項(xiàng)目，可能會(huì)對(duì)你有幫助，源代碼地址：gitee.com/hanxt/zimug-monitor-threadpool

經(jīng)過(guò)我的測(cè)試，放入orderDelayQueue的延時(shí)任務(wù)，在半小時(shí)之后得到正確的執(zhí)行處理。說(shuō)明我們的實(shí)現(xiàn)是正確的。

四、優(yōu)缺點(diǎn)

• 使用DelayQueue實(shí)現(xiàn)延時(shí)任務(wù)非常簡(jiǎn)單，而且簡(jiǎn)便，全部都是標(biāo)準(zhǔn)的JDK代碼實(shí)現(xiàn)，不用引入第三方依賴（不依賴redis實(shí)現(xiàn)、消息隊(duì)列實(shí)現(xiàn)等），非常的輕量級(jí)。
• 它的缺點(diǎn)就是所有的操作都是基于應(yīng)用內(nèi)存的，一旦出現(xiàn)應(yīng)用單點(diǎn)故障，可能會(huì)造成延時(shí)任務(wù)數(shù)據(jù)的丟失。如果訂單并發(fā)量非常大，因?yàn)镈elayQueue是無(wú)界的，訂單量越大，隊(duì)列內(nèi)的對(duì)象就越多，可能造成OOM的風(fēng)險(xiǎn)。所以使用DelayQueue實(shí)現(xiàn)延時(shí)任務(wù)，只適用于任務(wù)量較小的情況。

到此這篇關(guān)于java通過(guò)DelayQueue實(shí)現(xiàn)延時(shí)任務(wù)的文章就介紹到這了,更多相關(guān)java DelayQueue延時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論