如何使用 Redis 的訊息佇列

Redis 只是提供了乙個高效能的、原子操作的分布式佇列實現。 具體業務還是需要自己定製的。

你的需求實際上是一種畸形的生產者-消費者滿足。 對於這種需求，最主要的是將請求與實際處理過程解耦，一般採取非同步方式通知請求者，這與是否使用Redis關係不大。 一般的實現方式是，需要將使用者的請求封裝成乙個任務，然後將任務推送到Redis佇列中，然後後端就可以完全多程序、多執行緒併發處理任務，並將處理結果傳送給請求者。

這裡唯一的問題是任務需要設計為能夠包含請求資訊（請求內容、請求者 ID 等）。

可以使用 Redis 的列表型別，具體實現可以參考以下 **。

this->redis->rpush($key, $val);插入 - 直接插入。

this->redis->lpop($key);消費 - 被排除在外。

我目前的系統以這種方式穩定地工作。

訊息佇列應該能夠支援元件通訊訊息的快速讀寫，而Redis本身支援資料的高速訪問，可以滿足訊息佇列的讀寫效能要求。 此外，訪問訊息時必須滿足三個要求

在本節中，我們將分析Redis實現訊息佇列的解決方案。

blpop：當隊列為空時被阻止。

lpush：佇列已滿時阻塞。

brpoplpush：檢索消耗量並將其儲存到另乙個備份列表

從訊息有序性、唯一性、可靠性三個方面分析是否可行。

為了解決可靠性問題，可以使用 brpoplpush

使用者恢復後，可以從備份佇列中檢索訊息進行處理。

streams 是 Redis 專門為訊息佇列設計的一種資料型別，它提供了多種訊息佇列操作鏈命令。

xadd：插入一條訊息，訊息的格式為鍵值對形式，保證了順序，並能自動生成全域性唯一ID;

xread：用於讀取訊息，可以通過 ID 讀取資料;

xreadgroup：以消費者組的形式讀取訊息。

xpending：用於查詢每個使用組中所有使用者已讀取但尚未確認的訊息。

xack：用於向訊息佇列確認訊息處理已完成。

-over

通常使用列表來實現佇列操作，所以有很小的限制，所以任務是統一的先進先出，如果想要對某個任務進行優先順序排序並不容易處理，這就需要讓漏洞佇列有優先順序的概念，我們可以對高階任務進行優先順序排序，實現方式如下：

1）單列表實現：佇列的正常操作是左進右出（lpush、rpop）為了先處理高優先順序的任務，遇到高優先順序任務時，可以直接切入佇列，直接放到佇列頭（rpush）中，這樣當你從佇列頭（右）拿到任務時， 您將獲得高優先順序任務 （RPOP）。

2）使用兩個佇列，乙個普通佇列，乙個高階佇列，根據任務車的級別放置不同的佇列，獲取任務也很簡單，redis的brpop命令可以按順序從多個佇列中獲取值，brpop會按照給定鍵的順序檢視，並在找到的第乙個非空列表的末尾彈出乙個元素， redis> brpop list1 list2 0

list1 用作高優先順序任務的佇列，list2 用作普通任務的佇列。

這樣可以先處理高優先順序任務，當沒有高優先順序任務時，轉到正常任務。

方法1是最簡單的，但實際應用比較有限，模式3可以達到複雜的優先順序，但實現比較複雜，不利於維護，方法2是推薦用法，實際應用最合適。

1. 故事背景

1、讀取離線檔案資料，然後通過【離線資料】作為條件查詢第三方介面，返回最終結果，然後進入倉庫。

2.業務邏輯非常簡單讀取檔案，查詢介面，返回資料集，儲存資料庫。

3. 業務特性：第三方介面呼叫為 400 毫秒 （ms）。

如果用乙個普通的單執行緒執行乙個請求500毫秒，一天會執行8W以上的資料，超過20億個資料要到猴年才能執行。

二、治療方案

a） 初始方案使用 ganymed-ssh2（檔案儲存在 Linux 伺服器上）讀取檔案，使用 Redis 儲存訊息和多執行緒來提高處理能力。

b） 流程圖：

3. 問題的陳述

四、優化問題

最終流程圖：

通過 redis 製作計數器每次讀取一行以記錄值時，即使在服務終止後，也會首先從 Redis 讀取源值。

然後使用 CAT 指定的行數開始讀取資料。

將金鑰分片分割槽到不同的小金鑰儲存，減輕了單個節點的儲存壓力，充分利用了儲存資源。

3、redis push提供批處理模式（leftpushall），您可以指定讀取行數，然後批量入庫，而 POP 不會一次只提供乙個 POP 批處理。

4. 消費者通過多執行緒 pop 和重新分發到執行緒進行處理。

5.總結問題

關於Redis是什麼，我就不多說了，我在網上搜尋了很多文章。

首先，讓我們談談我試圖實現的內容：

類似乙個明確回答的褲子純資訊中轉站，如果有人想發訊息，先把訊息發給我，然後我進行**，為了有統一管理和修改的方便，並且所有的資訊都有優先權，也會有定時的激發純智慧傳送（如果同時傳送的訊息太多， 會有延遲）。

思路：首先，乙個是將這兩個佇列分成兩個佇列，乙個用於實現訊息優先順序，另乙個用於實現定時傳送。

使用有序的 Redis 集，當與 zadd 相加時，將分數與優先順序進行比較，也可以使用時間戳作為分數，用於表示時間。

PHP版本易於實現。

將郵件新增到優先順序佇列，並將 1 和 2 替換為在計畫時間傳送佇列的時間。

1 $redis = new redis();

2 $redis->connect('', 6379);

3 $redis->zadd('zset1', 1, 'message');

4 $redis->zadd('zset1', 2, 'message2');

將資料從佇列中拉出。

1 $redis->zrevrangebyscore('zset1, 'inf', inf', array('withscores'=>false, 'limit'=>array(0,20)))

此語句表示，如果要從小到大使用 zrangebyscore，則從 zset1 佇列中從最大 （+inf） 到最小 （-inf） 從 20 個條目中獲取，而不使用 score

如果你想在命令列上執行所有這些，你可以參考以下內容，當然已經被刪掉了。

1 connect('', 6379);

9 $redis->zrevrangebyscore('zset1', inf', inf', array('withscores'=>false, 'limit'=>array(0,20)))

10 exit;

11 } else

pcntl fork 是 PHP 中生成的子程序，呼叫函式時會返回乙個程序 pid，當 pid 為 0 時，表示它在子程序中，所以把所有要執行的東西都放在這裡。

上線的乙個專案，執行了幾個月，子流程方法沒有問題，也沒有掛起，這還算不錯。

基於Redis訊息佇列 - 簡訊服務。

常用的訊息佇列有rabbitmq、activemq，我個人認為這種訊息佇列太大太重，本文介紹基於Redis的輕量級訊息佇列服務。 一般來說，訊息佇列有兩種模式，一種是發布者訂閱模式，另一種是生產者和消費者模式。 Redis 的訊息佇列也是基於這兩個原則的隱式實現。

生產者將訊息放入佇列中，多個消費者一起監聽，誰先搶到資源，誰就從佇列中取出訊息進行處理。 請注意，每條訊息最多只能由乙個使用者接收。 <>

一般來說，訊息佇列有兩種型別：發布者-訂閱者模式和生產者-消費者模式。 訊息佇列可以在兩種場景中使用 high Redis 和 high Redis 實現。 定義：

生產者-消費者模式：生產者在佇列中產生訊息，多個消費者同時監聽佇列，誰先抓取訊息，誰就從佇列中獲取訊息。 也就是說，每條訊息最多只能由乙個使用者擁有。 發布者訂閱者模式：

有兩種方法可以做到這一點：

Redis 自帶 pub sub 機制，即 pub-subscribe 模型。 在這種模式下，生產者和消費者之間的關係是 1-M，即一條訊息會被多個消費者消費，當只有乙個消費者時，可以看作是 1-1 的訊息佇列，但這種方法不適合主體的場景。 首先，無法保證資料的可靠性，最終需要丟失主體的資料，如果訊息丟失，Redis宕機，部分資料不持久，甚至突然的網路抖動都可能帶來資料丟失，這應該是無法承受的。

如果前端寫入的資料過多，同步速度會變慢，資料不同步的時間越長，風險越大。 該方案更適合那些對資料可靠性要求不高的人，比如租用一些統計日誌。

Redis 的 push pop 機制利用了 Redis 列表資料結構。 乙個好的使用模型是生成 lpush 訊息、消費者 brpop 訊息，並設定超時時間以減輕對 Redis 的壓力。 與第一種方案相比，該方案提高了資料可靠性，只有在Redis宕機且資料不持久化時，才能根據業務情況通過AOF和縮短持久化間隔來保證資料的高可靠性，還可以通過多個客戶端提高消費速度。

但是，與專業的訊息佇列相比，該方案中訊息的狀態過於簡單（無狀態），並且沒有ACK機制，訊息取出後無法消費，取決於孫志釗的客戶端記錄日誌或推送回佇列。