【由淺至深】redis 實現發布訂閱的幾種方式


非常感謝依樂祝發表文章《.NET Core開發者的福音之玩轉Redis的又一傻瓜式神器推薦》,對csredis作了一次完整的詮釋。

前言

提到消息隊列,最熟悉無疑是 rabbitmq,它基本是業界標准的解決方案。本文詳細介紹 redis 多種實現輕訂閱方法,作者認為非常有趣並加以總結,希望對有需要的朋友學習 redis 功能有一定的帶入作用。

方法一:SUBSCRIBE + PUBLISH

//程序1:使用代碼實現訂閱端
var sub = RedisHelper.Subscribe(("chan1", msg => Console.WriteLine(msg.Body)));
//sub.Disponse(); //停止訂閱

//程序2:使用代碼實現發布端
RedisHelper.Publish("chan1", "111");

優勢:支持多端訂閱、簡單、性能高;
缺點:數據會丟失;

參考資料:http://doc.redisfans.com/pub_sub/subscribe.html

方法二:BLPOP + LPUSH(爭搶)

//程序1:使用代碼實現訂閱端
while (running) {
	try {
		var msg = RedisHelper.BLPop(5, "list1");
		if (string.IsNullOrEmpty(msg) == false) {
			Console.WriteLine(msg);
		}
	} catch (Exception ex) {
		Console.WriteLine(ex.Message);
	}
}

//程序2:使用代碼實現發布端
RedisHelper.LPush("list1", "111");

優勢:數據不會丟失、簡單、性能高;
缺點:不支持多端(存在資源爭搶);

總結:為了解決方法一的痛點,我們實現了本方法,並且很漂亮的制造了一個新問題(不支持多端訂閱)。

學習使用 BLPOP

BLPOP key [key ...] timeout

BLPOP 是列表的阻塞式(blocking)彈出原語。

它是 LPOP 命令的阻塞版本,當給定列表內沒有任何元素可供彈出的時候,連接將被 BLPOP 命令阻塞,直到等待超時或發現可彈出元素為止。

當給定多個 key 參數時,按參數 key 的先后順序依次檢查各個列表,彈出第一個非空列表的頭元素。

非阻塞行為

當 BLPOP 被調用時,如果給定 key 內至少有一個非空列表,那么彈出遇到的第一個非空列表的頭元素,並和被彈出元素所屬的列表的名字一起,組成結果返回給調用者。

當存在多個給定 key 時, BLPOP 按給定 key 參數排列的先后順序,依次檢查各個列表。

假設現在有 job 、 command 和 request 三個列表,其中 job 不存在, command 和 request 都持有非空列表。考慮以下命令:

BLPOP job command request 0

BLPOP 保證返回的元素來自 command ,因為它是按”查找 job -> 查找 command -> 查找 request “這樣的順序,第一個找到的非空列表。

redis> DEL job command request           # 確保key都被刪除
(integer) 0

redis> LPUSH command "update system..."  # 為command列表增加一個值
(integer) 1

redis> LPUSH request "visit page"        # 為request列表增加一個值
(integer) 1

redis> BLPOP job command request 0       # job 列表為空,被跳過,緊接着 command 列表的第一個元素被彈出。
1) "command"                             # 彈出元素所屬的列表
2) "update system..."                    # 彈出元素所屬的值

阻塞行為

如果所有給定 key 都不存在或包含空列表,那么 BLPOP 命令將阻塞連接,直到等待超時,或有另一個客戶端對給定 key 的任意一個執行 LPUSH 或 RPUSH 命令為止。

超時參數 timeout 接受一個以秒為單位的數字作為值。超時參數設為 0 表示阻塞時間可以無限期延長(block indefinitely) 。

redis> EXISTS job                # 確保兩個 key 都不存在
(integer) 0
redis> EXISTS command
(integer) 0

redis> BLPOP job command 300     # 因為key一開始不存在,所以操作會被阻塞,直到另一客戶端對 job 或者 command 列表進行 PUSH 操作。
1) "job"                         # 這里被 push 的是 job
2) "do my home work"             # 被彈出的值
(26.26s)                         # 等待的秒數

redis> BLPOP job command 5       # 等待超時的情況
(nil)
(5.66s)                          # 等待的秒數

更多學習資料:http://doc.redisfans.com/list/blpop.html

方法三:BLPOP + LPUSH(非爭搶)

本方法根據方法二演變而來,設計圖如下:

如何實現三端訂閱,都可收到消息,三端分別為 sub3, sub4, sub5:

1、sub3, sub4, sub5 使用【方法二】訂閱 listkey:list1_sub3,list1_sub4,list1_sub5;

2、總訂閱端訂閱 listkey:list1,總訂閱端收到消息后,執行 lpush list1_sub1 msg, lpush list1_sub2 msg, lpush list1_sub3 msg;

總訂閱端訂閱原始消息,隨后將消息分發給其他訂閱端,從而解決【方法二】不支持多端同時訂閱的缺點。

最終實現的邏輯為:多端先爭搶 list1 消息,搶到者再向其他端轉發消息。

測試代碼

nuget Install-Package CSRedisCore

var rds = new CSRedis.CSRedisClient("127.0.0.1:6379,password=,poolsize=50,ssl=false,writeBuffer=10240");

//sub1, sub2 爭搶訂閱(只可一端收到消息)
var sub1 = rds.SubscribeList("list1", msg => Console.WriteLine($"sub1 -> list1 : {msg}"));
var sub2 = rds.SubscribeList("list1", msg => Console.WriteLine($"sub2 -> list1 : {msg}"));

//sub3, sub4, sub5 非爭搶訂閱(多端都可收到消息)
var sub3 = rds.SubscribeListBroadcast("list2", "sub3", msg => Console.WriteLine($"sub3 -> list2 : {msg}"));
var sub4 = rds.SubscribeListBroadcast("list2", "sub4", msg => Console.WriteLine($"sub4 -> list2 : {msg}"));
var sub5 = rds.SubscribeListBroadcast("list2", "sub5", msg => Console.WriteLine($"sub5 -> list2 : {msg}"));

//sub6 是redis自帶的普通訂閱
var sub6 = rds.Subscribe(("chan1", msg => Console.WriteLine(msg.Body)));

Console.ReadKey();
sub1.Dispose();
sub2.Dispose();
sub3.Dispose();
sub4.Dispose();
sub5.Dispose();
sub6.Dispose();

rds.Dispose();
return;

測試功能時,發布端可以使用 redis-cli 工具。

結語

redis 功能何其多且相當好玩有趣 ,大家應盡可能多帶着興趣愛好去學習它。

若文中有不好的地方,請提出批評與改正方法,謝謝觀賞。

本文使用到 CSRedisCore 的開源地址:https://github.com/2881099/csredis


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM