当前位置：网站首页>Redis實現消息隊列的4種方案

Redis實現消息隊列的4種方案

2022-06-12 01:40:00 【柔情柴少】

redis實現消息隊列的集中方案：
1、基於List的LPUSH和BRPOP實現
2、PUB/SUB，訂閱/發布模式
3、基於Sorted-Set的實現
4、基於Stream類型的實現

1、基於list列錶的lpush和brpop實現消息隊列

生產者，模擬30個商品id，存入list列錶

<?php
$redis = new Redis();
$redis->connect('127.0.0.1',6379);
$redis->select(0);
$goodsId = 30;
$key = "sec_kill:active:active_1";
// 循環添加商品
for ($i=1;$i<=$goodsId;$i++){
    $redis->rPush($key,$i);
}

消費者，把商品id從list列錶取出來，模擬用戶id，一一對應30對，存入hash，剩下2970用戶秒殺失敗，存一個數量，並且把失敗用戶的編號存入列錶list

<?php
// 假裝是用戶的唯一標識
$uuid = md5(uniqid('user').time());
// 創建鏈接redis對象
$redis = new Redis();
$redis->connect('127.0.0.1',6379);
$redis->select(0);
$goodsKey = "sec_kill:active:active_1";
$orderKey = "sec_kill:active:order_1";
$failUserNum = "sec_kill:active:fail_user_num";
$failUser = "sec_kill:active:fail_user";
if ($goodsId = $redis->lPop($goodsKey)){
    // 秒殺成功
    // 把幸運用戶存在集合中
    $redis->hSet($orderKey,$goodsId,$uuid);
    echo "下單成功".PHP_EOL;
}else{
    // 秒殺失敗  將失敗用戶計數
    $redis->incr($failUserNum);
    $redis->rPush($failUser,$uuid);
    echo "下單失敗".PHP_EOL;
}

按理說，應該有監聽者watcher，監聽watcher的list,是否還有商品沒有消費，如果過期沒消費，則應該存入生產列錶，再次循環消費（博客觀看者，這一段可以不用看，作者寫給自己的，因為和上下文不對應，可能把你看迷糊）

2、redis發布/訂閱（PUB/SUB)實現消息隊列

（1）首先，介紹2組概念

channel：頻道，管道，信道，它就是一個通道，所有訂閱了該頻道的客戶端，都可以收到該頻道發出的消息

pattern：模式，或者說，多個頻道的集合，（有點類似正則的感覺），訂閱了該模式，所有能匹配到的頻道所發出的消息，該客戶端都能收到

（2）channel和pattern底層實現原理

當一個客戶端執行subscribe命令，訂閱某個或某些頻道的時候，這個客戶端於被訂閱頻道之間就建立起了一種訂閱關系。

Redis將所有頻道的訂閱關系都保存在服務器狀態的pubsub_channels字典裏面，這個字典的鍵是被訂閱的頻道，而鍵的值則是一個鏈錶，鏈錶裏面記錄了所有訂閱這個頻道的客戶端，底層C語言實現原理：


struct redisServer {
 
    // ...
 
    // 保存所有頻道的訂閱關系
    dict *pubsub_channels;
 
    // ...
 
};

根據頻道是否已有其他訂閱者，關聯操作分為兩種情况執行：
1、如果頻道已有其他訂閱者，那麼它在pubsub_channels字典中必然有相應的訂閱者鏈錶，程序唯一要做的就是將客戶端添加到訂閱者鏈錶的末尾
2、如果頻道還沒有任何訂閱者，那麼它必然不存在與pubsub_channels字典，程序首先要在pubsub_channels字典中為頻道創建一個鍵，並將這個鍵的值設置為空鏈錶，然後再將客戶端添加到鏈錶，成為鏈錶的第一個元素

subscribe命令的實現可以用以下偽代碼來描述：

def subscribe(*all_input_channels):
 
    # 遍曆輸入的所有頻道
    for channel in all_input_channels:
 
        # 如果 channel 不存在於 pubsub_channels 字典（沒有任何訂閱者）
        # 那麼在字典中添加 channel 鍵，並設置它的值為空鏈錶
        if channel not in server.pubsub_channels:
            server.pubsub_channels[channel] = []
 
        # 將訂閱者添加到頻道所對應的鏈錶的末尾
        server.pubsub_channels[channel].append(client)

聲明：這一部分借鑒了《RabbitMQ兔老大》，本文秉著友好交流學習的目的做一下個人總結，不做任何商用，原文連接：https://blog.csdn.net/hebtu666/article/details/114827837

unsubscribe命令和subscribe命令行為正好相反，當一個客戶端退訂某個或者某些頻道時，服務器將從pubsub_channels中解除客戶端於被退訂頻道之間的關聯：

1、程序會根據被退訂頻道的名字，在 pubsub_channels 字典中找到頻道對應的訂閱者鏈錶，然後從訂閱者鏈錶中删除退訂客戶端的信息
2、如果删除退訂客戶端之後，頻道的訂閱者鏈錶變成了空鏈錶，那麼說明這個頻道已經沒有任何訂閱者了，程序將從 pubsub_channels 字典中删除頻道對應的鍵

unsubscribe命令的實現可以用以下偽代碼來描述：


def unsubscribe(*all_input_channels):
 
    # 遍曆要退訂的所有頻道
    for channel in all_input_channels:
 
        # 在訂閱者鏈錶中删除退訂的客戶端
        server.pubsub_channels[channel].remove(client)
 
        # 如果頻道已經沒有任何訂閱者了（訂閱者鏈錶為空）
        # 那麼將頻道從字典中删除
        if len(server.pubsub_channels[channel]) == 0:
            server.pubsub_channels.remove(channel)

模式部分：

服務器將所有模式的訂閱者關系存在了pubsub_Patterns屬性裏


struct redisServer {
 
    // 保存所有頻道的訂閱關系
    list *pubsub_patterns;
 
};

pubsub_Patterns屬性是一個鏈錶，每個結點是被訂閱的模式，節點內記錄了模式，節點內的client屬性記錄了訂閱模式的客戶端


typedef struct pubsubPattern{
    //訂閱模式的客戶端
    redisClient *client;
    //被訂閱的模式
    robj *pattern;
}pubsubPattern;

每當客戶端執行PSUBSCRIBE這個命令來訂閱某個或某些模式時，服務器會對每個被訂閱的模式執行下面的操作：

1）新建一個pubsubPattern結構，設置好兩個屬性

2）將新節點加到pubsub_patterns尾部

偽代碼實現：


def osubscribe(*all_input_patterns):
    #遍曆所有輸入的模式
    #記錄被訂閱的模式和對應的客戶端
    pubsubPattern=create()
    pubsubPattern.client=client
    pubsubPattern.pattern=pattern
 
    #插入鏈錶末尾
    server.pub_patterns.append(pubsubPattern)

模式退訂命令PUNSUBSCRIBE是PSUBSCRIBE的反操作

服務器將找到並删除那些被退訂的模式


def osubscribe(*all_input_patterns):
    #遍曆所有退訂的模式
    for pattern in all_input_patterns:
        #遍曆每一個節點
        for pubsubPattern in server.pubsub_patterns:
            #如果客戶端和模式都相同
            if client==pubsubPattern.client:
                if pattern==pubsubPattern.pattern:
                    #删除
                    server.pub_patterns.remove(pubsubPattern)

發送消息：

當一個客戶端執行PUBLISH<channel> <message>命令將消息發送給頻道時，服務器需要：

1）把消息發送給所有本頻道的訂閱者

具體做法就是去pubsub_channels字典找到本頻道的鏈錶，也就是訂閱名單，然後發消息

2）將消息發給，包含本頻道的所有模式中的所有訂閱者

具體做法就是去pubsub_patterns查找包含本頻道的模式，並且把消息發送給訂閱它們的客戶端。

查看訂閱信息：

redis2.8新增三個命令，用來查看頻道和模式的相關信息。

PUBLISH CHANNELS[pattern]用於返回服務器當前被訂閱的頻道，pattern可寫可不寫，不寫就查看所有，否則查看與pattern匹配的對應頻道

這個子命令是通過遍曆pubsub_channels字典實現的。

PUBLISH NUMSUB[CHANNEL-1 CHANNEL-2.....]返回這些頻道的訂閱者數量

這個子命令是通過遍曆pubsub_channels字典，查看對應鏈錶長度實現的。