Java如何處理延遲任務(wù)過程解析
1、利用延遲隊列
延時隊列，第一他是個隊列，所以具有對列功能第二就是延時，這就是延時對列，功能也就是將任務(wù)放在該延時對列中，只有到了延時時刻才能從該延時對列中獲取任務(wù)否則獲取不到……
應(yīng)用場景比較多，比如延時1分鐘發(fā)短信，延時1分鐘再次執(zhí)行等，下面先看看延時隊列demo之后再看延時隊列在項(xiàng)目中的使用：
簡單的延時隊列要有三部分：第一實(shí)現(xiàn)了Delayed接口的消息體、第二消費(fèi)消息的消費(fèi)者、第三存放消息的延時隊列，那下面就來看看延時隊列demo。
一、消息體
package com.delqueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/**
* 消息體定義 實(shí)現(xiàn)Delayed接口就是實(shí)現(xiàn)兩個方法即compareTo 和 getDelay最重要的就是getDelay方法，這個方法用來判斷是否到期…… */
public class Message implements Delayed {
private int id;
private String body; // 消息內(nèi)容
private long excuteTime;// 延遲時長，這個是必須的屬性因?yàn)橐凑者@個判斷延時時長。
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();
}
// 自定義實(shí)現(xiàn)比較方法返回 1 0 -1三個參數(shù)
@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);
}
// 延遲任務(wù)是否到時就是按照這個方法判斷如果返回的是負(fù)數(shù)則說明到期否則還沒到期
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(this.excuteTime - System.nanoTime(), TimeUnit.NANOSECONDS);
}
}
二、消息消費(fèi)者
package com.delqueue;
import java.util.concurrent.DelayQueue;
public class Consumer implements Runnable {
// 延時隊列 ,消費(fèi)者從其中獲取消息進(jìn)行消費(fèi)
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("消費(fèi)消息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);
// 啟動消費(fèi)線程 消費(fèi)添加到延時隊列中的消息，前提是任務(wù)到了延期時間
ExecutorService exec = Executors.newFixedThreadPool(1);
exec.execute(new Consumer(queue));
exec.shutdown();
}
}
將消息體放入延遲隊列中，在啟動消費(fèi)者線程去消費(fèi)延遲隊列中的消息，如果延遲隊列中的消息到了延遲時間則可以從中取出消息否則無法取出消息也就無法消費(fèi)。
這就是延遲隊列demo，下面我們來說說在真實(shí)環(huán)境下的使用。
使用場景描述：
在打車軟件中對訂單進(jìn)行派單的流程，當(dāng)有訂單的時候給該訂單篩選司機(jī)，然后給當(dāng)訂單綁定司機(jī)，但是有時運(yùn)氣沒那么好，訂單進(jìn)來后第一次沒有篩選到合適的司機(jī)，但我們也不能就此結(jié)束派單，而是將該訂單的信息放到延時隊列中過個2秒鐘在進(jìn)行一次，其實(shí)這個2秒鐘就是一個延遲，所以這里我們就可以使用延時隊列來實(shí)現(xiàn)……
下面看看簡單的流程圖：
下面來看看具體代碼實(shí)現(xiàn)：
在項(xiàng)目中有如下幾個類：第一 、任務(wù)類 第二、按照任務(wù)類組裝的消息體類 第三、延遲隊列管理類
任務(wù)類即執(zhí)行篩選司機(jī)、綁單、push消息的任務(wù)類
package com.test.delayqueue;
/**
* 具體執(zhí)行相關(guān)業(yè)務(wù)的業(yè)務(wù)類
* @author whd
* @date 2017年9月25日 上午12:49:32
*/
public class DelayOrderWorker implements Runnable {
@Override
public void run() {
// TODO Auto-generated method stub
//相關(guān)業(yè)務(wù)邏輯處理
System.out.println(Thread.currentThread().getName()+" do something ……");
}
}
消息體類，在延時隊列中這個實(shí)現(xiàn)了Delayed接口的消息類是比不可少的，實(shí)現(xiàn)接口時有一個getDelay(TimeUnit unit)方法，這個方法就是判斷是否到期的
這里定義的是一個泛型類，所以可以將我們上面的任務(wù)類作為其中的task，這樣就將任務(wù)類分裝成了一個消息體
package com.test.delayqueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/**
* 延時隊列中的消息體將任務(wù)封裝為消息體
*
* @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; // 任務(wù)類，也就是之前定義的任務(wù)類
/**
* @param timeout
* 超時時間(秒)
* @param task
* 任務(wù)
*/
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;
}
}
延時隊列管理類，這個類主要就是將任務(wù)類封裝成消息并并添加到延時隊列中，以及輪詢延時隊列從中取出到時的消息體，在獲取任務(wù)類放到線程池中執(zhí)行任務(wù)
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;
/**
* 延時隊列管理類，用來添加任務(wù)、執(zhí)行任務(wù)
*
* @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;
// 守護(hù)線程
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("當(dāng)前存活線程數(shù)量:" + map.size());
int taskNum = delayQueue.size();
System.out.println("當(dāng)前延時任務(wù)數(shù)量:" + taskNum);
try {
// 從延時隊列中獲取任務(wù)
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();
}
}
}
/**
* 添加任務(wù)
*
* @param task
* @param time
* 延時時間
* @param unit
* 時間單位
*/
public void put(Runnable task, long time, TimeUnit unit) {
// 獲取延時時間
long timeout = TimeUnit.NANOSECONDS.convert(time, unit);
// 將任務(wù)封裝成實(shí)現(xiàn)Delayed接口的消息體
DelayOrderTask<?> delayOrder = new DelayOrderTask<>(timeout, task);
// 將消息體放到延時隊列中
delayQueue.put(delayOrder);
}
/**
* 刪除任務(wù)
*
* @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();// 任務(wù)1
DelayOrderWorker work2 = new DelayOrderWorker();// 任務(wù)2
DelayOrderWorker work3 = new DelayOrderWorker();// 任務(wù)3
// 延遲隊列管理類，將任務(wù)轉(zhuǎn)化消息體并將消息體放入延遲對列中等待執(zhí)行
DelayOrderQueueManager manager = DelayOrderQueueManager.getInstance();
manager.put(work1, 3000, TimeUnit.MILLISECONDS);
manager.put(work2, 6000, TimeUnit.MILLISECONDS);
manager.put(work3, 9000, TimeUnit.MILLISECONDS);
}
}
OK 這就是項(xiàng)目中的具體使用情況，當(dāng)然具體內(nèi)容被忽略，整體框架就是這樣，還有這里使用java的延時隊列但是這種方式是有問題的如果如果down機(jī)則會出現(xiàn)任務(wù)丟失，所以也可以考慮使用mq、redis來實(shí)現(xiàn)
2、mq實(shí)現(xiàn)延遲消息
在rabbitmq 3.5.7及以上的版本提供了一個插件（rabbitmq-delayed-message-exchange）來實(shí)現(xiàn)延遲隊列功能。同時插件依賴Erlang/OPT 18.0及以上。
插件源碼地址：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange
插件下載地址：
https://bintray.com/rabbitmq/community-plugins/rabbitmq_delayed_message_exchange
安裝：
進(jìn)入插件安裝目錄
{rabbitmq-server}/plugins/（可以查看一下當(dāng)前已存在的插件）
下載插件
rabbitmq_delayed_message_exchange wget https://bintray.com/rabbitmq/community-plugins/download_file?file_path=rabbitmq_delayed_message_exchange-0.0.1.ez
（如果下載的文件名稱不規(guī)則就手動重命名一下如：
rabbitmq_delayed_message_exchange-0.0.1.ez）
啟用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
關(guān)閉插件
rabbitmq-plugins disable rabbitmq_delayed_message_exchange
插件使用
通過聲明一個x-delayed-message類型的exchange來使用delayed-messaging特性
x-delayed-message是插件提供的類型，并不是rabbitmq本身的，發(fā)送消息的時候通過在header添加”x-delay”參數(shù)來控制消息的延時時間
直接在maven工程的pom.xml文件中加入
接下來在 application.properties 文件中加入redis配置：
也很簡單，代碼如下：
實(shí)現(xiàn)消息發(fā)送
x-delay
在這里我設(shè)置的延遲時間是3秒。
消息消費(fèi)者
直接在main方法里運(yùn)行Spring Boot程序,Spring Boot會自動解析 MessageReceiver 類的。
接下來只需要用Junit運(yùn)行一下發(fā)送消息的接口即可。
運(yùn)行完后，可以看到如下信息：
消息發(fā)送時間:2018-05-03 12:44:53 3秒鐘后，Spring Boot控制臺會輸出: 消息接收時間:2018-05-03 12:44:56 接收到的消息:hello i am delay msg
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
SpringBoot整合Elasticsearch實(shí)現(xiàn)索引和文檔的操作方法
Elasticsearch 基于 Apache Lucene 構(gòu)建，采用 Java 編寫，并使用 Lucene 構(gòu)建索引、提供搜索功能,本文分步驟通過綜合案例給大家分享SpringBoot整合Elasticsearch的相關(guān)知識，感興趣的朋友跟隨小編一起看看吧2021-05-05
Spring?AI?+?混元帶你實(shí)現(xiàn)企業(yè)級穩(wěn)定可部署的AI業(yè)務(wù)智能體
我們深入探討了Spring?AI在智能體構(gòu)建中的實(shí)際應(yīng)用,特別是在企業(yè)環(huán)境中的價值與效能,通過逐步實(shí)現(xiàn)一個本地部署的智能體解決方案,我們不僅展示了Spring?AI的靈活性與易用性,還強(qiáng)調(diào)了它在推動AI技術(shù)與業(yè)務(wù)深度融合方面的潛力,感興趣的朋友一起看看吧2024-11-11
這篇文章主要介紹了淺談Java中hashCode的正確求值方法，分享了相關(guān)代碼示例，小編覺得還是挺不錯的，具有一定借鑒價值，需要的朋友可以參考下2018-02-02
java中for循環(huán)執(zhí)行的順序圖文詳析
關(guān)于java的for循環(huán)想必大家非常熟悉,它是java常用的語句之一,這篇文章主要給大家介紹了關(guān)于java中for循環(huán)執(zhí)行順序的相關(guān)資料,需要的朋友可以參考下2021-06-06
Idea Jrebel 報錯:Cannot reactivate,offline
本文主要介紹了Idea Jrebel 報錯:Cannot reactivate,offline seat in use，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06
MyBatis中使用foreach循環(huán)的坑及解決
這篇文章主要介紹了MyBatis中使用foreach循環(huán)的坑及解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01
Java實(shí)現(xiàn)摳圖片文字或簽名的完整代碼
這篇文章主要介紹了java摳圖片文字或簽名的運(yùn)行原理，本文分步驟通過實(shí)例代碼給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-06-06
SpringBoot使用maven指定依賴包的版本(解決示例)
我們在使用A依賴的時候,這個依賴有引入了第三方B依賴,這時候我想指定B依賴的版本號,下面?zhèn)€大家分享解決示例,對SpringBoot maven依賴包相關(guān)配置方法感興趣的朋友一起看看吧2024-04-04
本文介紹了如何創(chuàng)建一個自定義的Spring Boot Starter,以實(shí)現(xiàn)日志功能,通過使用SPI機(jī)制,可以在不修改啟動類的情況下,實(shí)現(xiàn)自動配置和功能導(dǎo)入,同時,還討論了如何在自定義Starter中編寫必要的配置文件和注解,以確保功能的正確實(shí)現(xiàn)和配置的智能提示2025-02-02
作為初步進(jìn)入java開發(fā)學(xué)習(xí)的小白來說,學(xué)習(xí)java語言一開始的時候得一步步的學(xué)習(xí),比如說java輸入語句應(yīng)該這么去實(shí)現(xiàn)呢,這篇文章主要給大家介紹了關(guān)于Java輸入語句的相關(guān)資料,需要的朋友可以參考下2024-03-03