Redis實現延遲隊列的方案總結
作者：龍蝦
Redis實現的延遲隊列適用于處理一些比較簡單的業務，如發送郵件、發送通知等，對于復雜的業務不適用于Redis的延遲任務方案。?
redis是我們項目開發中常見的技術中間件，它除了可以實現常見的分布式鎖和分布式緩存功能之外，還可以幫助我們實現很多的功能，如延遲隊列。下面介紹幾種redis常見的實現延遲隊列的方案。
1、通過過期key通知實現
圖片
實現思路：首先開啟redis的key過期通知，然后在業務中給key設置過期時間，到了過期時間后redis會自動的將過期的key消息推送給監聽者，從而實現延遲任務。
核心的代碼實現：
#1、開始redis的過期通知
notify-keyspace-events Ex
#2、監聽redis的過期key
@Component
@Slf4j
public class RedisExpireKeyService extends
KeyExpirationEventMessageListener {
/**
* Creates new {@link MessageListener} for {@code __keyevent@*__:expired} messages.
*
* @param listenerContainer must not be {@literal null}.
*/
public RedisExpireKeyService(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
/**
* 監聽過期的key
*
*/
@Override
public void onMessage(Message message, byte[] pattern) {
String expireKey = message.toString();
//執行具體的業務
System.out.println("監聽到key=" + expireKey + "，已經過期");
}
}生產環境是不推薦使用此方案，原因Redis 的過期策略采用的是惰性刪除和定期刪除相結合的方式，redis并不保證 key 在過期時會被立即刪除操作，此方案適用于平時自己項目練習的時候使用。
2、通過Zset數據類型+定時任務實現
圖片
實現思路：ZSet 是一種有序集合類型，它可以存儲不重復的元素，并且給每個元素賦予一個 double 類型的排序權重值(score)，所以可以將元素的過期時間作為分值，通過定時任務掃描的方式判斷是否達到過期時間，從而實現延遲隊列。
核心的代碼實現：
#使用xxl-job
@JobName("consumerTaskJob")
public void consumerTaskJob() {
String expireKey = "ExPIRE_KEY";
try {
//獲取當前時間
double currentTime = System.currentTimeMillis();
//獲取超時的數據
Set<String> expiredMemberSet = redisTemplate.opsForZSet().rangeByScore(expireKey, Double.MIN_VALUE, currentTime);
//過期key
for (String expiredMember : expiredMemberSet) {
//todo 做實際的延遲任務
//從ZSet中移除數據
redisTemplate.opsForZSet().remove(expireKey, expiredMember);
}
} catch (Exception e) {
log.error("數據處理失敗",e);
}
}Zset+定時任務的實現延遲任務的方式雖然比監聽過期key方案合理一些，但是它還是存在一定的缺陷，如無重試機制、延遲時間固定化（依賴定時任務的執行時間）、不適用于大規模的延遲任務。
3、Redisson實現延遲隊列
Redisson是一個操作Redis的 Java 客戶框架，它提供了RDelayedQueue 接口和 RQueue 接口可以實現延遲隊列（Redisson 提供的延遲隊列底層也是基于 Zset 數據結構實現的）。
核心的代碼實現：
#1、添加依賴
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.0</version>
</dependency>
#2、添加數據到隊列中
//創建RedissonClient實例
RedissonClient redissonClient = Redisson.create();
//創建阻塞隊列
RBlockingDeque<String> queue = redissonClient.getBlockingDeque("delayQueue");
//創建延遲隊列并關聯到阻塞隊列
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);
//添加延遲任務
delayedQueue.offer("Task1", 5000, TimeUnit.MILLISECONDS);
#3、消費數據
while (true) {
try {
//獲取并移除隊首元素，如果隊列為空，則阻塞等待
String task = queue.take();
System.out.println("Task: " + task);
} catch (Exception e) {
log.error("消費失敗",e);
}
}總結：Redis實現的延遲隊列適用于處理一些比較簡單的業務，如發送郵件、發送通知等，對于復雜的業務不適用于Redis的延遲任務方案。
責任編輯：武曉燕
來源：
龍蝦編程
