delay-queue

redis實現延遲消息隊列

需求背景

最近在做一個排隊取號的系統

在用戶預約時間到達前XX分鐘發短信通知

在用戶預約時間結束時要判斷用戶是否去取號了，不然就記錄為爽約

在用戶取號后開始，等待XX分鐘后要發短信提醒是否需要使用其他渠道辦理

類似的場景太多，最簡單的解決辦法就是定時任務去掃表。這樣每個業務都要維護自己的掃表邏輯，

而且隨著時間的推移數據量會越來越多的，有的數據可能會延遲比較大

經過一番搜索，網上說rabbitmq可以滿足延遲執行需求，但是目前系統用了其他消息中間件，所以不打算用。

基于Redis實現的延遲消息隊列java版：項目github地址：delay-queue

整體結構

整個延遲隊列由4個部分組成：

JobPool用來存放所有Job的元信息。利用redis 的hash結構

DelayBucket是一組以時間為維度的有序隊列，用來存放所有需要延遲的Job(這里只存放Job Id)。利用redis 的 有序集合zset

Timer負責實時掃描各個Bucket，并將delay時間大于等于當前時間的Job放入到對應的Ready Queue。利用redis 的list 結構

ReadyQueue存放處于Ready狀態的Job(這里只存放JobId)，以供消費程序消費。

結構圖

消息結構

每個Job必須包含一下幾個屬性：

topic：Job類型。可以理解成具體的業務名稱。

id：Job的唯一標識。用來檢索和刪除指定的Job信息。

delayTime：jod延遲執行的時間，13位時間戳

ttr(time-to-run)：Job執行超時時間。單位：秒。主要是為了消息可靠性

message：Job的內容，供消費者做具體的業務處理，以json格式存儲。

舉例說明一個Job的生命周期

用戶預約后，同時往JobPool里put一個job。job結構為：{‘topic':'book’, ‘id':'123456’, ‘delayTime’:1517069375398 ,’ttrTime':60 , ‘message':’XXXXXXX’}

同時以jobId作為value，delayTime作為score 存到bucket 中，用jobId取模，放到10個bucket中，以提高效率

timer每時每刻都在輪詢各個bucket，按照score排序去最小的一個，當delayTime < 當前時間后，，取得job id從job pool中獲取元信息。

如果這時該job處于deleted狀態，則pass，繼續做輪詢；如果job處于非deleted狀態，首先再次確認元信息中delayTime是否大于等于當前時間，

如果滿足則根據topic將jobId放入對應的ready queue，然后從bucket中移除,并且；如果不滿足則重新計算delay時間，再次放入bucket，并將之前的job id從bucket中移除。

消費端輪詢對應的topic的ready queue，獲取job后做自己的業務邏輯。與此同時，服務端將已經被消費端獲取的job按照其設定的TTR，重新計算執行時間，并將其放入bucket。

消費端處理完業務后向服務端響應finish，服務端根據job id刪除對應的元信息。如果消費端在ttr時間內沒有響應，則ttr時間后會再收到該消息

后續擴展

加上超時重發次數

實現思路

任務job內容包含Array{0,0,2m,10m,10m,1h,2h,6h,15h}和通知到第幾次N(這里N=1, 即第1次).

消費者從隊列中取出任務, 根據N取得對應的時間間隔為0, 立即發送通知.

第1次通知失敗, N += 1 => 2

從Array中取得間隔時間為2m, 添加一個延遲時間為2m的任務到延遲隊列, 任務內容仍包含Array和N

第2次通知失敗, N += 1 => 3, 取出對應的間隔時間10m, 添加一個任務到延遲隊列, 同上

......

第7次通知失敗, N += 1 => 8, 取出對應的間隔時間15h, 添加一個任務到延遲隊列, 同上

第8次通知失敗, N += 1 => 9, 取不到間隔時間, 結束通知

引用說明

參考有贊延遲隊列思路設計實現

總結

以上是生活随笔為你收集整理的java redis延迟队列_基于redis实现的延迟消息队列的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。