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)致部分數(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高并發(fā)解決方案以及高負載優(yōu)化方法
這篇文章主要介紹了淺談Java高并發(fā)解決方案以及高負載優(yōu)化方法，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08
這篇文章主要介紹了如何搭建一個SpringBoot項目,包括環(huán)境準備、創(chuàng)建新項目、探索項目結(jié)構(gòu)以及展望未來,通過詳細的步驟和實用的技巧,幫助開發(fā)者快速上手SpringBoot開發(fā),文中通過圖文介紹的非常詳細,需要的朋友可以參考下2025-02-02
這篇文章主要介紹了使用@ControllerAdvice同時配置過濾多個包的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-06-06
這篇文章主要介紹了AndroidQ沙盒機制之分區(qū)存儲適配，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-06-06
今天給大家總結(jié)一下SpringBoot下攔截器的使用，需要的朋友參考下吧2017-05-05
jsoup 是一款 Java 的 HTML 解析器,它提供了一套非常便利的 API,可通過 DOM、CSS 通過類似于 JQuery 的操作方法來取出和操作數(shù)據(jù),這篇文章主要介紹了SpringBoot使用jsoup爬取HTML,需要的朋友可以參考下2024-02-02
SpringBoot的@Value給靜態(tài)變量注入application.properties屬性值
這篇文章主要介紹了SpringBoot的@Value給靜態(tài)變量注入application.properties屬性值,Spring是一個開源的框架，主要是用來簡化開發(fā)流程,通過IOC，依賴注入（DI）和面向接口實現(xiàn)松耦合,需要的朋友可以參考下2023-05-05
Springboot 如何實現(xiàn)filter攔截token驗證和跨域
這篇文章主要介紹了Springboot 如何實現(xiàn)filter攔截token驗證和跨域操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-08-08
Java代碼實現(xiàn)發(fā)送QQ和網(wǎng)易電子郵件
這篇文章主要為大家詳細介紹了如何通過Java代碼實現(xiàn)發(fā)送QQ和網(wǎng)易電子郵件功能,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2026-02-02
這篇文章主要介紹了Java 實現(xiàn)簡單Socket 通信的示例，幫助大家更好的理解和學(xué)習(xí)使用Java，感興趣的朋友可以了解下2021-03-03