前言

提到消息隊(duì)列，最熟悉無疑是 rabbitmq，它基本是業(yè)界標(biāo)準(zhǔn)的解決方案。本文詳細(xì)介紹 redis 多種實(shí)現(xiàn)輕訂閱方法，作者認(rèn)為非常有趣并加以總結(jié)，希望對(duì)有需要的朋友學(xué)習(xí) redis 功能有一定的帶入作用。

方法一：SUBSCRIBE + PUBLISH

image

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

//程序2：使用代碼實(shí)現(xiàn)發(fā)布端
RedisHelper.Publish("chan1", "111");

優(yōu)勢(shì)：支持多端訂閱、簡(jiǎn)單、性能高；
缺點(diǎn)：數(shù)據(jù)會(huì)丟失；

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

方法二：BLPOP + LPUSH（爭(zhēng)搶）

image

//程序1：使用代碼實(shí)現(xiàn)訂閱端
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：使用代碼實(shí)現(xiàn)發(fā)布端
RedisHelper.LPush("list1", "111");

優(yōu)勢(shì)：數(shù)據(jù)不會(huì)丟失、簡(jiǎn)單、性能高；
缺點(diǎn)：不支持多端（存在資源爭(zhēng)搶）；

總結(jié)：為了解決方法一的痛點(diǎn)，我們實(shí)現(xiàn)了本方法，并且很漂亮的制造了一個(gè)新問題（不支持多端訂閱）。

學(xué)習(xí)使用 BLPOP

BLPOP key [key ...] timeout

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

它是 LPOP 命令的阻塞版本，當(dāng)給定列表內(nèi)沒有任何元素可供彈出的時(shí)候，連接將被 BLPOP 命令阻塞，直到等待超時(shí)或發(fā)現(xiàn)可彈出元素為止。

當(dāng)給定多個(gè) key 參數(shù)時(shí)，按參數(shù) key 的先后順序依次檢查各個(gè)列表，彈出第一個(gè)非空列表的頭元素。

非阻塞行為

當(dāng) BLPOP 被調(diào)用時(shí)，如果給定 key 內(nèi)至少有一個(gè)非空列表，那么彈出遇到的第一個(gè)非空列表的頭元素，并和被彈出元素所屬的列表的名字一起，組成結(jié)果返回給調(diào)用者。

當(dāng)存在多個(gè)給定 key 時(shí)， BLPOP 按給定 key 參數(shù)排列的先后順序，依次檢查各個(gè)列表。

假設(shè)現(xiàn)在有 job 、 command 和 request 三個(gè)列表，其中 job 不存在， command 和 request 都持有非空列表?？紤]以下命令：

BLPOP job command request 0

BLPOP 保證返回的元素來自 command ，因?yàn)樗前础辈檎?job -> 查找 command -> 查找 request “這樣的順序，第一個(gè)找到的非空列表。

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

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

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

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

阻塞行為

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

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

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

redis> BLPOP job command 300     # 因?yàn)閗ey一開始不存在，所以操作會(huì)被阻塞，直到另一客戶端對(duì) job 或者 command 列表進(jìn)行 PUSH 操作。
1) "job"                         # 這里被 push 的是 job
2) "do my home work"             # 被彈出的值
(26.26s)                         # 等待的秒數(shù)

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

更多學(xué)習(xí)資料：http://doc.redisfans.com/list/blpop.html

方法三：BLPOP + LPUSH（非爭(zhēng)搶）

本方法根據(jù)方法二演變而來，設(shè)計(jì)圖如下：

image

如何實(shí)現(xiàn)三端訂閱，都可收到消息，三端分別為 sub3, sub4, sub5：

1、sub3, sub4, sub5 使用【方法二】訂閱 listkey：list1_sub3，list1_sub4，list1_sub5；

2、總訂閱端訂閱 listkey：list1，總訂閱端收到消息后，執(zhí)行 lpush list1_sub1 msg， lpush list1_sub2 msg， lpush list1_sub3 msg；

總訂閱端訂閱原始消息，隨后將消息分發(fā)給其他訂閱端，從而解決【方法二】不支持多端同時(shí)訂閱的缺點(diǎn)。

最終實(shí)現(xiàn)的邏輯為：多端先爭(zhēng)搶 list1 消息，搶到者再向其他端轉(zhuǎn)發(fā)消息。

測(cè)試代碼

nuget Install-Package CSRedisCore

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

//sub1, sub2 爭(zhēng)搶訂閱（只可一端收到消息）
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 非爭(zhēng)搶訂閱（多端都可收到消息）
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;

測(cè)試功能時(shí)，發(fā)布端可以使用 redis-cli 工具。

image

結(jié)語(yǔ)

redis 功能何其多且相當(dāng)好玩有趣 ，大家應(yīng)盡可能多帶著興趣愛好去學(xué)習(xí)它。

若文中有不好的地方，請(qǐng)?zhí)岢雠u(píng)與改正方法，謝謝觀賞。