Redis實現延遲隊列的方案總結

作者：龍蝦 2024-05-08 14:49:22

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的延遲任務方案。

責任編輯：武曉燕來源：龍蝦編程

Redis 延遲隊列業務