SpringBoot中Redisson延遲隊列的示例
更新時間：2024年10月16日 09:28:39 作者：Frank-fu
延時隊列是一種常見的需求,延時隊列允許我們延遲處理某些任務(wù),本文主要介紹了Redisson延遲隊列的示例,具有一定的參考價值,感興趣的可以了解一下
場景：
需求：
支付的二維碼，超過兩個小時以后，如果還未支付，則自動轉(zhuǎn)為取消支付，或者支付超時的狀態(tài)
需求分析：
1，動態(tài)定時任務(wù)：
每個支付的二維碼創(chuàng)建的時候，創(chuàng)建一個動態(tài)的定時任務(wù)，兩個小時候自動執(zhí)行，更新支付狀態(tài)，可以解決這個問題。
（1）持久化：
如果服務(wù)重啟了，動態(tài)定時任務(wù)會丟失，導(dǎo)致部分?jǐn)?shù)據(jù)沒辦法更新狀態(tài)。
（2）分布式：
如果當(dāng)服務(wù)重啟時，自動掃描數(shù)據(jù)，重新計算時間，再次創(chuàng)建動態(tài)定時任務(wù)。可以解決（1）的問題，但是當(dāng)分布式，多個節(jié)點的時候，都會重新加載所有的任務(wù)，這樣性能上不是最優(yōu)解，只能在數(shù)據(jù)源上加上節(jié)點名稱，不同的服務(wù)節(jié)點，加載屬于自己的定時任務(wù)，可以解決這個問題?？偟南胂?，太麻煩了，還是算了。
2，Redisson延遲隊列
（1）持久化：隊列信息放在Redis上，服務(wù)重啟不影響。
（2）分布式：多節(jié)點去Redis拿去數(shù)據(jù)，誰搶到算誰的，不會存在同一個任務(wù)，多個節(jié)點支持。唯一不足就是過度依賴Redis，萬一Redis崩了，那就涼涼了（那就是要把Redis配置高可用，當(dāng)前業(yè)務(wù)就不用管了）?？傮w來說還是比較好用的。
實現(xiàn)
1，創(chuàng)建延遲隊列的監(jiān)聽任務(wù)【RedisDelayedQueueListener】，消費延遲隊列
2，創(chuàng)建新增延遲隊列的類，用于創(chuàng)建延遲隊列
3，整體初始化，把監(jiān)聽任務(wù)與spring綁定，掃描各個監(jiān)聽延遲隊列的實現(xiàn)類，并開啟單獨線程，監(jiān)聽任務(wù)。
4，創(chuàng)建延遲任務(wù)（開始測試使用）
連接Redis
不貼代碼了，自己在網(wǎng)上搜
監(jiān)聽延遲隊列
接口：
/**
* 隊列事件監(jiān)聽接口，需要實現(xiàn)這個方法
*
* @module
* @author frank
* @date 2021/8/19 10:50
*/
public interface RedisDelayedQueueListener<T> {
/**
* 執(zhí)行方法
*
* @param t
*/
void invoke(T t);
}實現(xiàn)：
import com.sxmaps.netschool.common.redisson.RedisDelayedQueueListener;
import com.sxmaps.netschool.service.vo.school.SchoolAccountPayStateReqVO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 支付二維碼監(jiān)聽器
*
* @module
* @author frank
* @date 2021/8/19 10:49
*/
@Component
public class PayQCordListener implements RedisDelayedQueueListener<SchoolAccountPayStateReqVO> {
private final Logger logger = LoggerFactory.getLogger(PayQCordListener.class);
@Autowired
private SchoolAccountService schoolAccountService;
@Override
public void invoke(SchoolAccountPayStateReqVO payStateReqVO) {
logger.info("支付二維碼-延遲失效,內(nèi)容:{}", payStateReqVO);
//處理業(yè)務(wù)，更新二維碼狀態(tài)
logger.info("支付二維碼-延遲失效,內(nèi)容:{},處理結(jié)果:{}", payStateReqVO,respDTO);
}
}增加延遲隊列
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* 增加延遲信息
*
* @author frank
* @module
* @date 2021/8/19 10:49
*/
@Component
public class RedisDelayedQueue {
private final Logger logger = LoggerFactory.getLogger(RedisDelayedQueue.class);
@Autowired
RedissonClient redissonClient;
/**
* 添加隊列
*
* @param t DTO傳輸類
* @param delay 時間數(shù)量
* @param timeUnit 時間單位
* @param <T> 泛型
*/
private <T> void addQueue(T t, long delay, TimeUnit timeUnit, String queueName) {
logger.info("添加延遲隊列,監(jiān)聽名稱:{},時間:{},時間單位:{},內(nèi)容:{}" , queueName, delay, timeUnit,t);
RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
delayedQueue.offer(t, delay, timeUnit);
}
/**
* 添加隊列-秒
*
* @param t DTO傳輸類
* @param delay 時間數(shù)量
* @param <T> 泛型
*/
public <T> void addQueueSeconds(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.SECONDS, clazz.getName());
}
/**
* 添加隊列-分
*
* @param t DTO傳輸類
* @param delay 時間數(shù)量
* @param <T> 泛型
*/
public <T> void addQueueMinutes(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.MINUTES, clazz.getName());
}
/**
* 添加隊列-時
*
* @param t DTO傳輸類
* @param delay 時間數(shù)量
* @param <T> 泛型
*/
public <T> void addQueueHours(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.HOURS, clazz.getName());
}
/**
* 添加隊列-天
*
* @param t DTO傳輸類
* @param delay 時間數(shù)量
* @param <T> 泛型
*/
public <T> void addQueueDays(T t, long delay, Class<? extends RedisDelayedQueueListener> clazz) {
addQueue(t, delay, TimeUnit.DAYS, clazz.getName());
}
}整體初始化
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import java.util.Map;
/**
* 初始化隊列監(jiān)聽
*
* @module
* @author frank
* @date 2021/8/19 10:49
*/
@Component
public class RedisDelayedQueueInit implements ApplicationContextAware {
private final Logger logger = LoggerFactory.getLogger(RedisDelayedQueueInit.class);
@Autowired
RedissonClient redissonClient;
/**
* 獲取應(yīng)用上下文并獲取相應(yīng)的接口實現(xiàn)類
*
* @param applicationContext
* @throws BeansException
*/
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
for (Map.Entry<String, RedisDelayedQueueListener> taskEventListenerEntry : map.entrySet()) {
String listenerName = taskEventListenerEntry.getValue().getClass().getName();
startThread(listenerName, taskEventListenerEntry.getValue());
}
}
/**
* 啟動線程獲取隊列*
*
* @param queueName queueName
* @param redisDelayedQueueListener 任務(wù)回調(diào)監(jiān)聽
* @param <T> 泛型
* @return
*/
private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
//服務(wù)重啟后，無offer，take不到信息。
redissonClient.getDelayedQueue(blockingFairQueue);
//由于此線程需要常駐，可以新建線程，不用交給線程池管理
Thread thread = new Thread(() -> {
logger.info("啟動監(jiān)聽隊列線程" + queueName);
while (true) {
try {
T t = blockingFairQueue.take();
logger.info("監(jiān)聽隊列線程，監(jiān)聽名稱：{},內(nèi)容:{}", queueName, t);
redisDelayedQueueListener.invoke(t);
} catch (Exception e) {
logger.info("監(jiān)聽隊列線程錯誤,", e);
}
}
});
thread.setName(queueName);
thread.start();
}
}創(chuàng)建延遲任務(wù)
@Autowired RedisDelayedQueue queue; ................. queue.addQueueHours(new SchoolAccountPayStateReqVO(dto.getPayNo()),2, PayQCordListener.class);
到此這篇關(guān)于Redisson延遲隊列的示例的文章就介紹到這了,更多相關(guān)Redisson延遲隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Java基礎(chǔ)入門 Swing中間容器的使用，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
使用spring?data的page和pageable如何實現(xiàn)分頁查詢
這篇文章主要介紹了使用spring?data的page和pageable如何實現(xiàn)分頁查詢，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
spring基于注解配置實現(xiàn)事務(wù)控制操作
這篇文章主要介紹了spring基于注解配置實現(xiàn)事務(wù)控制操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
這篇文章主要介紹了Javamail使用過程中常見問題解決方案,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-08-08
Struts2攔截器Interceptor的原理與配置實例詳解
攔截器是一種AOP(面向切面編程)思想的編程方式.它提供一種機制是開發(fā)者能夠把相對獨立的代碼抽離出來,配置到Action前后執(zhí)行。下面這篇文章主要給大家介紹了關(guān)于Struts2攔截器Interceptor的原理與配置的相關(guān)資料，需要的朋友可以參考下。2017-11-11
Java利用Geotools從DEM數(shù)據(jù)中讀取指定位置的高程信息全過程
Geotools作為一款功能強大且開源的地理工具庫,為地理數(shù)據(jù)的處理和分析提供了豐富的類庫和便捷的接口,能夠很好地滿足從DEM數(shù)據(jù)中讀取高程信息這一實戰(zhàn)需求,本文將深入講解如何利用Geotools從獲取DEM數(shù)據(jù)到成功讀取指定位置高程信息的全過程,需要的朋友可以參考下2025-03-03
Spring Boot 使用 Swagger 構(gòu)建 RestAPI 接口文檔
這篇文章主要介紹了Spring Boot 使用 Swagger 構(gòu)建 RestAPI 接口文檔，幫助大家更好的理解和使用Spring Boot框架，感興趣的朋友可以了解下2020-10-10
這篇文章主要介紹了Java8 中 Map 接口的新方法,本文通過代碼實例給大家介紹的非常詳細(xì)，具有一定的參考借鑒價值，需要的朋友可以參考下2021-08-08
舉例解析Java的設(shè)計模式編程中里氏替換原則的意義
這篇文章主要介紹了Java的設(shè)計模式中里氏替換原則的意義,文中舉例來說明里氏替換原則中強調(diào)的繼承特性方面可能帶來的問題,需要的朋友可以參考下2016-02-02
這篇文章主要介紹了Java中HashMap的初始容量設(shè)置方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06