RM新时代网站-首页

0
  • 聊天消息
  • 系統消息
  • 評論與回復
登錄后你可以
  • 下載海量資料
  • 學習在線課程
  • 觀看技術視頻
  • 寫文章/發(fā)帖/加入社區(qū)
會員中心
創(chuàng)作中心

完善資料讓更多小伙伴認識你,還能領取20積分哦,立即完善>

3天內不再提示

Redis 延時隊列,一次性搞明白

數據分析與開發(fā) ? 來源:數據分析與開發(fā) ? 作者:數據分析與開發(fā) ? 2020-10-30 16:34 ? 次閱讀

所謂延時隊列就是延時的消息隊列,下面說一下一些業(yè)務場景

實踐場景

訂單支付失敗,每隔一段時間提醒用戶

用戶并發(fā)量的情況,可以延時2分鐘給用戶發(fā)短信

先來看看Redis實現普通的消息隊列

我們知道,對于專業(yè)的消息隊列中間件,如Kafka和RabbitMQ,消費者在消費消息之前要進行一系列的繁瑣過程。

如RabbitMQ發(fā)消息之前要創(chuàng)建 Exchange,再創(chuàng)建 Queue,還要將 Queue 和 Exchange 通過某種規(guī)則綁定起來,發(fā)消息的時候要指定 routingkey,還要控制頭部信息

但是絕大 多數情況下,雖然我們的消息隊列只有一組消費者,但還是需要經歷上面一些過程。

有了 Redis,對于那些只有一組消費者的消息隊列,使用 Redis 就可以非常輕松的搞定。Redis 的消息隊列不是專業(yè)的消息隊列,它沒有非常多的高級特性, 沒有 ack 保證,如果對消息的可靠性有著極致的追求,那么它就不適合使用

異步消息隊列基本實現

Redis 的 list(列表) 數據結構常用來作為異步消息隊列使用,使用 rpush/lpush 操作入隊列, 使用 lpop 和 rpop 來出隊列

>rpushqueue月伴飛魚1月伴飛魚2月伴飛魚3 (integer)3 >lpopqueue "月伴飛魚1" >llenqueue (integer)2

問題1:如果隊列空了

客戶端是通過隊列的 pop 操作來獲取消息,然后進行處理。處理完了再接著獲取消息, 再進行處理。如此循環(huán)往復,這便是作為隊列消費者的客戶端的生命周期。

可是如果隊列空了,客戶端就會陷入 pop 的死循環(huán),不停地 pop,沒有數據,接著再 pop, 又沒有數據。這就是浪費生命的空輪詢??蛰喸儾坏吡丝蛻舳说?CPU,redis 的 QPS 也 會被拉高,如果這樣空輪詢的客戶端有幾十來個,Redis 的慢查詢可能會顯著增多。

通常我們使用 sleep 來解決這個問題,讓線程睡一會,睡個 1s 鐘就可以了。不但客戶端 的 CPU 能降下來,Redis 的 QPS 也降下來了

問題2:隊列延遲

用上面睡眠的辦法可以解決問題。同時如果只有 1 個消費者,那么這個延遲就是 1s。如果有多個消費者,這個延遲會有所下降,因 為每個消費者的睡覺時間是岔開來的。

有沒有什么辦法能顯著降低延遲呢?

那就是 blpop/brpop。

這兩個指令的前綴字符 b 代表的是 blocking,也就是阻塞讀。

阻塞讀在隊列沒有數據的時候,會立即進入休眠狀態(tài),一旦數據到來,則立刻醒過來。消 息的延遲幾乎為零。用 blpop/brpop 替代前面的 lpop/rpop,就完美解決了上面的問題。

問題3:空閑連接自動斷開

其實他還有個問題需要解決—— 空閑連接的問題。

如果線程一直阻塞在哪里,Redis 的客戶端連接就成了閑置連接,閑置過久,服務器一般 會主動斷開連接,減少閑置資源占用。這個時候 blpop/brpop 會拋出異常來。

所以編寫客戶端消費者的時候要小心,注意捕獲異常,還要重試。

分布式鎖沖突處理

假如客戶端在處理請求時加分布式鎖沒加成功怎么辦。

一般有 3 種策略來處理加鎖失敗:

1、直接拋出異常,通知用戶稍后重試;

2、sleep 一會再重試;

3、將請求轉移至延時隊列,過一會再試;

直接拋出特定類型的異常

這種方式比較適合由用戶直接發(fā)起的請求,用戶看到錯誤對話框后,會先閱讀對話框的內 容,再點擊重試,這樣就可以起到人工延時的效果。如果考慮到用戶體驗,可以由前端的代碼 替代用戶自己來進行延時重試控制。它本質上是對當前請求的放棄,由用戶決定是否重新發(fā)起 新的請求。

sleep

sleep 會阻塞當前的消息處理線程,會導致隊列的后續(xù)消息處理出現延遲。如果碰撞的比 較頻繁或者隊列里消息比較多,sleep 可能并不合適。如果因為個別死鎖的 key 導致加鎖不成 功,線程會徹底堵死,導致后續(xù)消息永遠得不到及時處理。

延時隊列

這種方式比較適合異步消息處理,將當前沖突的請求扔到另一個隊列延后處理以避開沖突。

延時隊列的實現

我們可以使用 zset這個命令,用設置好的時間戳作為score進行排序,使用zadd score1 value1 ....命令就可以一直往內存中生產消息。再利用 zrangebysocre 查詢符合條件的所有待處理的任務,通過循環(huán)執(zhí)行隊列任務即可。也可以通過zrangebyscore key min max withscores limit 0 1查詢最早的一條任務,來進行消費

privateJedisjedis; publicvoidredisDelayQueueTest(){ Stringkey="delay_queue"; //實際開發(fā)建議使用業(yè)務ID和隨機生成的唯一ID作為value,隨機生成的唯一ID可以保證消息的唯一性,業(yè)務ID可以避免value攜帶的信息過多 StringorderId1=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId1); StringorderId12=UUID.randomUUID().toString(); jedis.zadd(queueKey,System.currentTimeMillis()+5000,orderId2); newThread(){ @Override publicvoidrun(){ while(true){ SetresultList; //只獲取第一條數據,只獲取不會移除數據 resultList=jedis.zrangebyscore(key,System.currentTimeMillis(),0,1); if(resultList.size()==0){ try{ Thread.sleep(1000); }catch(InterruptedExceptione){ e.printStackTrace(); break; } }else{ //移除數據獲取到的數據 if(jedis.zrem(key,resultList.get(0))>0){ StringorderId=resultList.get(0); log.info("orderId={}",resultList.get(0)); this.handleMsg(orderId); } } } } }.start(); } publicvoidhandleMsg(Tmsg){ System.out.println(msg); }

上面的實現, 在多線程邏輯上也是沒有問題的, 假設有兩個線程 T1, T2和其他更多線程, 處理邏輯如下, 保證了多線程情況下只有一個線程處理了對應的消息:

1.T1, T2 和其他更多線程調用 zrangebyscore 獲取到了一條消息 A

2.T1 準備開始刪除消息 A, 由于是原子操作, T2 和其他更多線程等待 T1 執(zhí)行 zrem 刪除消息 A 后再執(zhí)行 zrem 刪除消息 A

3.T1 刪除了消息 A, 返回刪除成功標記 1, 并對消息 A 進行處理

4.T2 其他更多線程開始 zrem 刪除消息 A, 由于消息 A 已經被刪除, 所以所有的刪除均失敗, 放棄了對消息 A 的處理

同時,我們要注意一定要對handle_msg進行異常捕獲,避免因為個別任務處理問題導致循環(huán)異常退 出

進一步優(yōu)化

上面的算法中同一個任務可能會被多個進程取到之后再使用 zrem 進行爭搶,那些沒搶到 的進程都是白取了一次任務,這是浪費??梢钥紤]使用 lua scripting 來優(yōu)化一下這個邏輯,將 zrangebyscore 和 zrem 一同挪到服務器端進行原子化操作,這樣多個進程之間爭搶任務時就不 會出現這種浪費了

使用調用Lua腳本進一步優(yōu)化

Lua 腳本, 如果有超時的消息, 就刪除, 并返回這條消息, 否則返回空字符串:

StringluaScript="localresultArray=redis.call('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1) "+ "if#resultArray>0then "+ "ifredis.call('zrem',KEYS[1],resultArray[1])>0then "+ "returnresultArray[1] "+ "else "+ "return'' "+ "end "+ "else "+ "return'' "+ "end"; jedis.eval(luaScript,ScriptOutputType.VALUE,newString[]{key},String.valueOf(System.currentTimeMillis()));

Redis延時隊列優(yōu)勢

Redis用來進行實現延時隊列是具有這些優(yōu)勢的:

1.Redis zset支持高性能的 score 排序。

2.Redis是在內存上進行操作的,速度非常快。

3.Redis可以搭建集群,當消息很多時候,我們可以用集群來提高消息處理的速度,提高可用性。

4.Redis具有持久化機制,當出現故障的時候,可以通過AOF和RDB方式來對數據進行恢復,保證了數據的可靠性

Redis延時隊列劣勢

使用 Redis 實現的延時消息隊列也存在數據持久化, 消息可靠性的問題

沒有重試機制 - 處理消息出現異常沒有重試機制, 這些需要自己去實現, 包括重試次數的實現等

沒有 ACK 機制 - 例如在獲取消息并已經刪除了消息情況下, 正在處理消息的時候客戶端崩潰了, 這條正在處理的這些消息就會丟失, MQ 是需要明確的返回一個值給 MQ 才會認為這個消息是被正確的消費了

如果對消息可靠性要求較高, 推薦使用 MQ 來實現

Redission實現延時隊列

基于Redis的Redisson分布式延遲隊列結構的RDelayedQueue Java對象在實現了RQueue接口的基礎上提供了向隊列按要求延遲添加項目的功能。該功能可以用來實現消息傳送延遲按幾何增長或幾何衰減的發(fā)送策略

RQueuedistinationQueue=... RDelayedQueuedelayedQueue=getDelayedQueue(distinationQueue); //10秒鐘以后將消息發(fā)送到指定隊列 delayedQueue.offer("msg1",10,TimeUnit.SECONDS); //一分鐘以后將消息發(fā)送到指定隊列 delayedQueue.offer("msg2",1,TimeUnit.MINUTES);

在該對象不再需要的情況下,應該主動銷毀。僅在相關的Redisson對象也需要關閉的時候可以不用主動銷毀。

RDelayedQueuedelayedQueue=... delayedQueue.destroy();

是不是很方便...............

責任編輯:xj

原文標題:Redis 延時隊列,這次徹底給你整明白了

文章出處:【微信公眾號:數據分析與開發(fā)】歡迎添加關注!文章轉載請注明出處。

聲明:本文內容及配圖由入駐作者撰寫或者入駐合作網站授權轉載。文章觀點僅代表作者本人,不代表電子發(fā)燒友網立場。文章及其配圖僅供工程師學習之用,如有內容侵權或者其他違規(guī)問題,請聯系本站處理。 舉報投訴
  • 編程
    +關注

    關注

    88

    文章

    3614

    瀏覽量

    93685
  • 線程
    +關注

    關注

    0

    文章

    504

    瀏覽量

    19675
  • Redis
    +關注

    關注

    0

    文章

    374

    瀏覽量

    10871

原文標題:Redis 延時隊列,這次徹底給你整明白了

文章出處:【微信號:DBDevs,微信公眾號:數據分析與開發(fā)】歡迎添加關注!文章轉載請注明出處。

收藏 人收藏

    評論

    相關推薦

    ADS1256第一次上電的時候,采集的ADC信號是實際值的半,為什么?

    會幾個小時復現一次,有時候會幾天復現一次,真的明白哪里出問題,希望各位大神幫忙解決下。具體會在哪些地方出問題。
    發(fā)表于 12-13 15:33

    智芯傳感ZXP4系列一次性血壓傳感器介紹

    智芯傳感ZXP4系列一次性血壓傳感器,主要供醫(yī)療單位對患者進行動脈壓、中心靜脈壓、肺動脈壓、左冠狀動脈壓等多種血壓監(jiān)測,屬于有創(chuàng)血壓監(jiān)測。
    的頭像 發(fā)表于 12-04 13:52 ?635次閱讀
    智芯傳感ZXP4系列<b class='flag-5'>一次性</b>血壓傳感器介紹

    【AI技術支持】USB_CDC電腦串口一次性發(fā)送100000byte丟包問題處理

    啟明云端/01你是否曾遇到過?在使用ESP32-S3,ESP-IDF版本為idf5.2時,蒸汽鍋產品基于例程tusb_serial_device測試USBCDC自發(fā)自收,電腦CDC串口一次性發(fā)送
    的頭像 發(fā)表于 11-07 08:01 ?267次閱讀
    【AI技術支持】USB_CDC電腦串口<b class='flag-5'>一次性</b>發(fā)送100000byte丟包問題處理

    一次電源與二電源有什么不同

    在電力系統和電子設備的供電領域中,一次電源與二電源是兩個至關重要的概念。它們各自承擔著不同的功能和角色,共同確保電力供應的穩(wěn)定性和可靠。本文將對一次電源與二
    的頭像 發(fā)表于 10-10 14:10 ?1735次閱讀

    ODU MEDI-SNAP一次性醫(yī)用插拔自鎖插頭產品介紹

    為滿足一次性內窺鏡、一次性手術消融刀等設備中的耗材需求,歐度全新推出了MEDI-SNAP一次性醫(yī)用插拔自鎖插頭,為醫(yī)療客戶打造了組在品質與經濟
    的頭像 發(fā)表于 09-10 09:59 ?433次閱讀

    aP8942A一次性編程(OTP)語音集成電路英文手冊

    AP8942A是一次性編程(OTP)語音集成電路(IC),它設計用于在各種應用中存儲和播放語音消息。以下是AP8942A語音IC的詳細介紹:核心特性存儲容量:內嵌1M bits的EPROM,可以
    發(fā)表于 08-08 14:40 ?4次下載

    AP8921A一次性編程(OTP)語音集成電路英文手冊

    AP8921A是一次性編程(OTP)語音集成電路(IC),它具備多種功能,適用于多種需要語音提示或聲音播放的應用場景。以下是AP8921A語音IC的詳細介紹:核心特性存儲容量:內嵌512k
    發(fā)表于 08-08 14:39 ?0次下載

    esp32如何一次性讀取大文件數據?

    esp32沒有提供數據庫讀寫的例子,最近有個大文件,無法一次性讀出,請問,怎么讀取,json中部分json數組。然后修改完了以后,在寫入進去?
    發(fā)表于 06-25 06:52

    一次性注射針剛性測試儀作用與重要

    文章由濟南三泉智能科技有限公司提供一次性注射針剛性測試儀是用于評估一次性注射針剛性性能的專用設備。、作用與重要確保注射針質量:注射針的剛性是評價其質量的重要參數之
    的頭像 發(fā)表于 06-04 15:32 ?323次閱讀
    <b class='flag-5'>一次性</b>注射針剛性測試儀作用與重要<b class='flag-5'>性</b>

    請問如何將TIM1_CCR1H和TIM1_CCR1L合并成個變量,之后一次性賦值?

    請問:TIM1_CCR1H和TIM1_CCR1L分別是比較寄存器的高八位和第八位,用來設置PWM的占空比的。是否可以把這兩個寄存器變成個整體,之后一次性對其進行賦值?
    發(fā)表于 05-10 07:47

    一次性使用無菌腦積水分流器綜合測試儀

    一次性使用無菌腦積水分流器綜合測試儀 描述 一次性使用無菌腦積水分流器是種用于治療腦積水的醫(yī)療器械。腦積水是指在顱內或腦脊液循環(huán)系統中積聚過多液體導致顱內壓力增高的病癥。腦積水分流
    的頭像 發(fā)表于 04-06 09:13 ?331次閱讀
    <b class='flag-5'>一次性</b>使用無菌腦積水分流器綜合<b class='flag-5'>性</b>測試儀

    九齊語音芯片NY5P025B SOP8 SOP16 OTP一次性燒錄IC方案開發(fā)

    九齊語音芯片NY5P025B SOP8 SOP16 OTP一次性燒錄IC方案開發(fā) 、引言 隨著電子科技的不斷發(fā)展,語音芯片在各種電子產品中的應用越來越廣泛。九齊科技的NY5P025B
    的頭像 發(fā)表于 03-13 14:39 ?946次閱讀

    GB 15811一次性醫(yī)用注射針檢測儀器

    GB 15811一次性醫(yī)用注射針檢測儀器
    發(fā)表于 02-18 14:21 ?0次下載

    nuc980從機一次性需要把許多數據發(fā)給上位機,每發(fā)個packet必須長時間延時,是什么原因導致的?

    , .unbind = hid_unbind, }; 把高速改成全速,但是好像沒起到作用,設備信息里還是高速。請問,從機一次性需要把許多數據發(fā)給上位機,每發(fā)個packet必須長時間延時,這是什么原因導致的。
    發(fā)表于 01-17 06:26

    一次性保險絲與自恢復保險絲這兩種之間有什么差別?

    一次性保險絲與自恢復保險絲這兩種之間有什么差別? 一次性保險絲和自恢復保險絲是兩種用于電氣系統中的保險器件,它們在功能和特性上存在些差異。本文將對這兩種保險絲進行比較,并詳細解釋它們的定義
    的頭像 發(fā)表于 01-04 14:13 ?1672次閱讀
    RM新时代网站-首页