2. 程式人生 > 實用技巧 >12. Redis中的訊息佇列

12. Redis中的訊息佇列

阿新 • • 發佈：2020-07-18

楔子

Redis雖然是一個快取，但是它也可以作為一個訊息佇列。所以redis還是比較有野心的，本來在快取方面就已經把memcached給幹掉了，但還想在訊息佇列的方向上闖一闖。不過雖說Redis支援訊息佇列，但是它還是作為快取更加的專業，大公司很少有將redis作為訊息佇列來使用的，因此訊息佇列的話一般還是使用rabbitmq、activemq之類的會比較好。

釋出訂閱模式

在 Redis 中提供了專門的型別：Publisher（釋出者）和 Subscriber（訂閱者）來實現訊息佇列。

不過在介紹訊息佇列之前，先丟擲幾個概念，這樣理解下文會更加輕鬆一些，當然都是老生常談的內容了。

  • 釋出訊息的叫做釋出方或釋出者，也就是訊息的生產者。
  • 接收訊息的叫做訊息的訂閱方或訂閱者，也就是消費者，用來處理生產者釋出的訊息。

除了釋出和和訂閱者，在訊息佇列中還有一個重要的概念：channel，指的是管道，可以理解為某個訊息佇列的名稱。首先消費者先要訂閱某個 channel，然後當生產者把訊息傳送到這個 channel 中時，消費者就可以正常接收到訊息了，如下圖所示：

普通訂閱與釋出

訊息佇列有兩個重要的角色，一個是傳送者，另一個就是訂閱者，對應的命令如下：

  • 釋出訊息：publish channel "message"
  • 訂閱訊息：subscribe channel

下面我們來看具體的命令實現。

訂閱訊息

127.0.0.1:6379> subscribe channel1 channel2  # 可以同時訂閱多個頻道
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel1"
3) (integer) 1
1) "subscribe"
2) "channel2"
3) (integer) 2

注意：當我們訂閱某個頻道的時候，就阻塞在這裡了。

就類似於微信公眾號一樣，你關注了某個公眾號，那麼當公眾號上面發表文章的時候，你就可以收到。此時操作公眾號的人就是訊息釋出者，你就是訊息訂閱者，公眾號就是訊息佇列，往公眾號上面發表的文章就是訊息。

傳送訊息

我們上面的訂閱者在訂閱之後，就處於阻塞狀態，因此我們需要再開一個終端。

127.0.0.1:6379> publish channel1 "mea: please please money"
(integer) 1
127.0.0.1:6379> publish channel2 "mea: please please money"
(integer) 1
127.0.0.1:6379>

返回值表示成功傳送給了幾個訂閱方，所以這裡的 1 就表示成功發給了一個訂閱者，這個數字可以是 0~n，這是由訂閱者的數量決定的。如果有兩個訂閱者，那麼返回值就是2。

然後我們來看看訂閱者：

127.0.0.1:6379> subscribe channel1 channel2
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel1"
3) (integer) 1
1) "subscribe"
2) "channel2"
3) (integer) 2
1) "message"
2) "channel1"  # channel1 接收到訊息
3) "mea: please please money"
1) "message"
2) "channel2"  # channel2接收到訊息
3) "mea: please please money"

主題訂閱

主題訂閱說白了，和模糊匹配是類似的。假設我們需要訂閱好幾個訊息佇列，但它們都是以log開頭的，那麼我們就可以通過psubscribe log*來自動訂閱所有以log開頭的佇列。

比如我們上面的channel1、channel2，我們就可以通過psubscribe channel*實現，至於訊息釋出者則不需要變。

當然主題訂閱也可以是多個，比如：psubscribe log* db*，訂閱所有以log開頭、db開頭的訊息佇列。

Python操作Redis的釋出訂閱

# 訂閱者
import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")

# 呼叫pubsub方法返回一個訂閱者
sub = client.pubsub()
# 訂閱兩個佇列
sub.subscribe("ch1", "ch2")
# 監聽，此時處於阻塞狀態
for item in sub.listen():
    # 一旦釋出者釋出訊息，這裡就可以接收到
    # item["channel"]是頻道，item["data"]是接收到了內容
    print(item["channel"], item["data"])

# 釋出者
import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")
# 釋出者很簡單，直接釋出訊息介面
client.publish("ch1", "屑女僕1")
client.publish("ch1", "屑女僕2")
client.publish("ch2", "屑女僕3")

當執行釋出者的時候，會發現訂閱者多了幾條輸出，至於內容顯然是釋出者釋出的內容。

Python操作Redis，訂閱者還有幾種方式。

import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")

sub = client.pubsub()
sub.subscribe("ch1", "ch2")

while True:
    # 這種方式會瞬間返回，如果有訊息得到訊息，沒有訊息會返回None
    item = sub.get_message()
    if item:
        print(item["channel"], item["data"])

或者開啟一個新的執行緒去監聽。

import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")

sub = client.pubsub()
sub.subscribe("ch1", "ch2")

def handler(item):
    print(item["channel"], item["data"])

# 給每一個頻道註冊一個處理函式，當頻道有訊息時，會自動將訊息傳遞給處理函式
# 注意：上面的pubsub中訂閱的頻道都要有對應的處理函式
# 假設我們只給ch1註冊了處理函式，那麼執行的時候就會報錯：Channel: 'ch2' has no handler registered
sub.channels.update({"ch1": handler, "ch2": handler})
# 開啟一個執行緒執行，會返回新開啟的執行緒物件，注意：因為是單獨開了一個執行緒，所以這裡不會阻塞的，會直接往下走
th = sub.run_in_thread()
print("xxx")
print("yyy")
print("zzz")

# 先啟動訂閱者，再啟動釋出者，程式輸出如下
"""
xxx
yyy
zzz
ch1 屑女僕1
ch1 屑女僕2
ch2 屑女僕3
"""

# 注意：這裡程式依舊會卡住，因為開啟的執行緒是非守護執行緒
# 所以即便主執行緒執行完畢，也依舊會等待子執行緒
# 解決的辦法有兩種：
# 一種是在run_in_thread中加上一個引數daemon=True，設定為守護執行緒，這樣主執行緒就不會等待了
# 另一種是手動停止，我們說sub.run_in_thread會返回新開啟的執行緒，然後呼叫其stop方法即可
th.stop()  # 通過這種方式，我們可以在任意時刻停止監聽。

對於主題訂閱，釋出者程式碼不用變，只需要將訂閱的sub.subscribe換成sub.psubscribe即可。

import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")

sub = client.pubsub()
sub.psubscribe("ch*")

def handler(item):
    print(item["channel"], item["data"])

# 對於開啟新的執行緒去監聽，要將之前的self.channels換成self.patterns
sub.patterns.update({"ch*": handler})
sub.run_in_thread()

取消訂閱

既然有訂閱，那麼就要取消訂閱，就類似於取關(o(╥﹏╥)o)。

使用unsubscribe channel1 channel2可以取消訂閱多個channel，同理對於psubscribe ch*，也有punsubscribe ch*取消訂閱指定模式的頻道。比較簡單，不再贅述。

注意事項

釋出訂閱模式存在以下兩個缺點：

  • 無法持久化儲存訊息，如果 Redis 伺服器宕機或重啟，那麼所有的訊息將會丟失；
  • 釋出訂閱模式是"發後既忘"的工作模式，如果有訂閱者離線重連之後不能消費之前的歷史訊息。

然而這些缺點在 Redis 5.0 添加了 Stream 型別之後會被徹底的解決。

除了以上缺點外，釋出訂閱模式還有另一個需要注意問題：當消費端有一定的訊息積壓時，也就是生產者傳送的訊息，消費者消費不過來時，如果超過 32M 或者是 60s 內持續保持在 8M 以上，消費端會被強行斷開，這個引數是在配置檔案中設定的，預設值是 client-output-buffer-limit pubsub 32mb 8mb 60。

小結

這一節介紹了訊息佇列的幾個名詞，生產者、消費者對應的就是訊息的傳送者和接收者，也介紹了釋出訂閱模式的幾個命令：

  • subscribe channel：普通訂閱
  • publish channel message：訊息推送
  • psubscribe pattern：主題訂閱
  • unsubscribe channel：取消普通訂閱
  • punsubscribe pattern：取消主題訂閱

使用它們之後就可以完成單個頻道和多個頻道的訊息收發，但傳送與訂閱模式也有一些缺點，比如“發後既忘”和不能持久化等問題，然而這些問題會等到 Stream 型別的出現而得到解決，關於更多 Stream 的內容後面文章會詳細介紹。

實現訊息佇列的其它方式

在 Redis 5.0 之前訊息佇列的實現方式有很多種，比較常見的除了我們上文介紹的釋出訂閱模式，還有兩種：List 和 ZSet 的實現方式。

List 和 ZSet 的方式解決了釋出訂閱模式不能持久化的問題，但這兩種方式也有自己的缺點，接下來我們一起來了解一下，先從 List 實現訊息佇列的方式說起。

List版訊息佇列

List 方式是實現訊息佇列最簡單和最直接的方式，它主要是通過 lpush 和 rpop 存入和讀取實現訊息佇列的，如下圖所示：

List 使用命令的方式實現訊息佇列：

127.0.0.1:6379> lpush channel message1
(integer) 1
127.0.0.1:6379> lpush channel message2
(integer) 2
127.0.0.1:6379> rpop channel
"message1"
127.0.0.1:6379> rpop channel
"message2"
127.0.0.1:6379>

lpush用於生產訊息，rpop用於消費訊息。

然後我們使用Python來操作List模擬訊息佇列。

import threading
import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")


def producer(messages: list):

    for message in messages:
        client.lpush("mq", message)
        print("生產者往佇列mq裡放入訊息:", message)


def consumer():
    while True:
        # brpop裡面可以傳入一個timeout，表示設定超時時間，預設為0，會一直阻塞
        print("消費者從佇列mq中消費了訊息:", client.brpop("mq"))


t1 = threading.Thread(target=producer, args=([f"message{_}" for _ in range(10)],))
t2 = threading.Thread(target=consumer)

t1.start()
t2.start()

for t in threading.enumerate():
    if t is not threading.main_thread():
        t.join()

"""
消費者從佇列mq中消費了訊息: ('mq', 'message0')
生產者往佇列mq裡放入訊息: message0
生產者往佇列mq裡放入訊息: message1
消費者從佇列mq中消費了訊息: ('mq', 'message1')
生產者往佇列mq裡放入訊息: message2
消費者從佇列mq中消費了訊息: ('mq', 'message2')
生產者往佇列mq裡放入訊息: message3
消費者從佇列mq中消費了訊息: ('mq', 'message3')
生產者往佇列mq裡放入訊息: message4
消費者從佇列mq中消費了訊息: ('mq', 'message4')
生產者往佇列mq裡放入訊息: message5
消費者從佇列mq中消費了訊息: ('mq', 'message5')
消費者從佇列mq中消費了訊息: ('mq', 'message6')
生產者往佇列mq裡放入訊息: message6
消費者從佇列mq中消費了訊息: ('mq', 'message7')
生產者往佇列mq裡放入訊息: message7
消費者從佇列mq中消費了訊息: ('mq', 'message8')
生產者往佇列mq裡放入訊息: message8
消費者從佇列mq中消費了訊息: ('mq', 'message9')
生產者往佇列mq裡放入訊息: message9
"""

我們看到使用List實現了一個類似於佇列的方式，但這顯然也是有其優缺點的

優點

  • 訊息可以被持久化，藉助 Redis 本身的持久化（AOF、RDB 或者是混合持久化），可以有效的儲存資料；
  • 消費者可以積壓訊息，不會因為客戶端的訊息過多而被強行斷開。

缺點

  • 訊息不能被重複消費，一個訊息消費完就會被刪除；
  • 沒有主題訂閱的功能。

ZSet 版訊息佇列

相比於之前的List 和釋出訂閱方式，ZSet 版訊息佇列在實現上要複雜一些，但 ZSet 因為多了一個 score（分值）屬性，從而使它具備更多的功能，例如我們可以用它來儲存時間戳，以此來實現延遲訊息佇列等。

它的實現思路和 List 相同也是利用 zadd 和 zrangebyscore 來實現存入和讀取，這裡就不重複敘述了，可以根據 List 的實現方式來實踐一下，看能不能實現相應的功能。如果寫不出來也沒關係，我們會在後面學習延遲佇列，到時候會用 ZSet 來實現。

優點

  • 支援訊息持久化；
  • 相比於 List 查詢更方便，ZSet 可以利用 score 屬性很方便的完成檢索，而 List 則需要遍歷整個元素才能檢索到某個值。

缺點

  • ZSet 不能儲存相同元素的值，也就是如果有訊息是重複的，那麼只能插入一條資訊在有序集合中；
  • ZSet 是根據 score 值排序的，不能像 List 一樣，按照插入順序來排序；
  • ZSet 沒有向 List 的 brpop 那樣的阻塞彈出的功能。

小結

這一節我們介紹了訊息佇列的另外兩種實現方式 List 和 ZSet，它們都是利用自身方法，先把資料放到佇列(自身的資料結構)裡，再使用無限迴圈讀取佇列中的訊息，以實現訊息佇列的功能，相比釋出訂閱模式，這兩種方式的優勢是支援持久化，當然它們各自都存在一些問題。

訊息佇列的終極解決方案--stream(上)

在 Redis 5.0 Stream 沒出來之前，訊息佇列的實現方式都有著各自的缺陷，例如：

  • 釋出訂閱模式 PubSub，不能持久化也就無法可靠的儲存訊息，並且對於離線重連的客戶端不能讀取歷史訊息的缺陷；
  • 列表實現訊息佇列的方式不能重複消費，一個訊息消費完就會被刪除；
  • 有序集合訊息佇列的實現方式不能儲存相同 value 的訊息，並且不能阻塞讀取訊息。

基礎使用

Stream 既然是一個數據型別，那麼和其他資料型別相似，它也有一些自己的操作方法，例如：

  • xadd：新增訊息；
  • xlen：查詢訊息的長度；
  • xdel：根據訊息ID刪除訊息；
  • del：刪除整個stream，當然del可以刪除任意的key；
  • xrange：讀取區間訊息；
  • xread：讀取某個訊息之後的訊息；

我們看一下具體如何操作。

新增訊息

語法：xadd key ID field1 string1 field2 string2······

127.0.0.1:6379> xadd my_stream * name mea age 19
"1594952987816-0"
127.0.0.1:6379>

其中*表示Redis使用的規則：時間戳+序號的方式自動生成ID，當然你也可以指定自己的ID

查詢訊息的長度

語法：xlen key

127.0.0.1:6379> xlen my_stream
(integer) 1
127.0.0.1:6379> xadd my_stream * name hanser age 28  # 再新增一條
"1594953077142-0"
127.0.0.1:6379> xlen my_stream  # 長度變為2
(integer) 2
127.0.0.1:6379>

刪除訊息

語法：xdel key 訊息ID·····，可以同時刪除多個

127.0.0.1:6379> xlen my_stream
(integer) 2
127.0.0.1:6379> xdel my_stream 1594953077142-0
(integer) 1
127.0.0.1:6379> xlen my_stream
(integer) 1
127.0.0.1:6379>

刪除整個stream

直接使用del，它可以刪除任意多個任意的key

127.0.0.1:6379> del my_stream
(integer) 1
127.0.0.1:6379>

查詢區間訊息

xrange key start end count n，這裡的start和end指的是訊息ID。

127.0.0.1:6379> # 新增幾條訊息
127.0.0.1:6379> xadd mq * name satori age 17
"1594953403230-0"
127.0.0.1:6379> xadd mq * name koishi age 16
"1594953410148-0"
127.0.0.1:6379> xadd mq * name scarlet age 400
"1594953422249-0"
127.0.0.1:6379> xadd mq * name morisa age unknow
"1594953438554-0"
127.0.0.1:6379> # 查詢
127.0.0.1:6379> xrange mq 1594953410148-0 1594953438554-0
1) 1) "1594953410148-0"
   2) 1) "name"
      2) "koishi"
      3) "age"
      4) "16"
2) 1) "1594953422249-0"
   2) 1) "name"
      2) "scarlet"
      3) "age"
      4) "400"
3) 1) "1594953438554-0"
   2) 1) "name"
      2) "morisa"
      3) "age"
      4) "unknow"
127.0.0.1:6379>
127.0.0.1:6379> # -表示第一條訊息、+表示最後一條訊息
127.0.0.1:6379> xrange mq - +
1) 1) "1594953403230-0"
   2) 1) "name"
      2) "satori"
      3) "age"
      4) "17"
2) 1) "1594953410148-0"
   2) 1) "name"
      2) "koishi"
      3) "age"
      4) "16"
3) 1) "1594953422249-0"
   2) 1) "name"
      2) "scarlet"
      3) "age"
      4) "400"
4) 1) "1594953438554-0"
   2) 1) "name"
      2) "morisa"
      3) "age"
      4) "unknow" 
127.0.0.1:6379> 
127.0.0.1:6379> # count n表示限定數量，這裡是返回兩條
127.0.0.1:6379> xrange mq - + count 2
1) 1) "1594953403230-0"
   2) 1) "name"
      2) "satori"
      3) "age"
      4) "17"
2) 1) "1594953410148-0"
   2) 1) "name"
      2) "koishi"
      3) "age"
      4) "16"
127.0.0.1:6379>

雖然這裡查詢用的是訊息ID，但是也要像索引一樣注意先後關係。start對應的訊息要在end對應的訊息之前，類似於索引。

查詢某個訊息之後的訊息

語法：xread count n streams xxx MESSAGE_ID

從名為xxx的stream中，讀取訊息ID為MESSAGE_ID的後n條訊息

127.0.0.1:6379> xread count 2 streams mq 1594953410148-0
1) 1) "mq"
   2) 1) 1) "1594953422249-0"
         2) 1) "name"
            2) "scarlet"
            3) "age"
            4) "400"
      2) 1) "1594953438554-0"
         2) 1) "name"
            2) "morisa"
            3) "age"
            4) "unknow"
127.0.0.1:6379>
127.0.0.1:6379> # 該訊息後面只剩一條訊息了，所以即便count為2，所以也只返回了一條
127.0.0.1:6379> xread count 2 streams mq 1594953422249-0
1) 1) "mq"
   2) 1) 1) "1594953438554-0"
         2) 1) "name"
            2) "morisa"
            3) "age"
            4) "unknow"
127.0.0.1:6379>

並且該命令還提供了一個可以阻塞讀取的引數block，我們可以使用它讀取某條資料之後的新增資料。

比如：xread count 1 block streams mq $

$表示最後一條，此時程式會阻塞，會一直讀取最後一條資料之後的新增資料，既然阻塞，那麼肯定要開啟兩個終端才會看得到現象。

127.0.0.1:6379> xread count 1 block 0 streams mq $  # 程式就卡在了這裡


127.0.0.1:6379> xadd mq * name mea age 19  # 新開一個視窗，新增資料
"1594969025661-0"
127.0.0.1:6379> 

127.0.0.1:6379> xread count 1 block 0 streams mq $
1) 1) "mq"
   2) 1) 1) "1594969025661-0"
         2) 1) "name"
            2) "mea"
            3) "age"
            4) "19"
(44.51s)
127.0.0.1:6379>  # 此時接收到了新新增的資料，另外此時監聽也就結束了。

Python實現stream

from pprint import pprint
import redis

client = redis.Redis(host="47.94.174.89", decode_responses="utf-8")


def producer():
    id_lst = []
    for _ in [
        {"name": "mashiro", "age": 17},
        {"name": "satori", "age": 17},
        {"name": "koishi", "age": 17}]:
        id_lst.append(client.xadd("ch", _))
    return id_lst


def consumer():
    id_lst = producer()
    # 第二個引數和第三個引數預設是"-"和"+"，也就是全部讀取
    # 當然我們也可以使用xread，具體引數可以看註釋
    msg = client.xrange("ch", id_lst[0], id_lst[-1])
    pprint(msg)


consumer()
"""
[('1594956361922-0', {'age': '17', 'name': 'mashiro'}),
 ('1594956361929-0', {'age': '17', 'name': 'satori'}),
 ('1594956361934-0', {'age': '17', 'name': 'koishi'})]
"""

訊息佇列的終極解決方案--stream(下)

下面我們使用訊息分組，不過在開始使用訊息分組之前，我們必須手動建立分組才行，以下是幾個和 Stream 分組有關的命令，我們先來學習一下它的使用。

訊息分組命令

建立消費者群組

語法：xgroup create <stream_key> <group_key> <ID>

127.0.0.1:6379> xgroup create mq group1 0-0
OK
127.0.0.1:6379>
  • mq：stream的key
  • group1：分組的名稱
  • 0-0：表示從第一條訊息開始讀取

如果從當前最後一條訊息向後讀取的話，那麼使用$即可。

127.0.0.1:6379> xgroup create mq group2 $
OK
127.0.0.1:6379>

讀取訊息

語法：xreadgroup group <group_key> <consumer_key> [count n] streams <stream_key>

  • group_key：建立的分組名
  • consumer_key：消費者名，隨便指定即可
  • count n：每次讀取的數量，可選，不指定全部返回
  • stream_key：佇列名稱
127.0.0.1:6379> xreadgroup group group1 c1 count 1 streams mq >  # 結尾應該還有個>,表示讀取下一條訊息
1) 1) "mq"
   2) 1) 1) "1594953403230-0"
         2) 1) "name"
            2) "satori"
            3) "age"
            4) "17"
127.0.0.1:6379> 
127.0.0.1:6379> xreadgroup group group1 古明地覺 count 1 streams mq >  # 消費者名字隨便起
1) 1) "mq"
   2) 1) 1) "1594953410148-0"
         2) 1) "name"
            2) "koishi"
            3) "age"
            4) "16"
127.0.0.1:6379>

這個引數類似於xread，也可以設定阻塞讀取。

127.0.0.1:6379> xreadgroup group group1 c2  streams mq >  # 不指定count，將訊息全部消費完
1) 1) "mq"
   2) 1) 1) "1594953422249-0"
         2) 1) "name"
            2) "scarlet"
            3) "age"
            4) "400"
      2) 1) "1594953438554-0"
         2) 1) "name"
            2) "morisa"
            3) "age"
            4) "unknow"
      3) 1) "1594969025661-0"
         2) 1) "name"
            2) "mea"
            3) "age"
            4) "19"
127.0.0.1:6379> xreadgroup group group1 c2 streams mq >  # 此時已經獲取不到訊息了
(nil)# 另外，我們這裡消費者數量是不受限制的，它們消費的都是同一個佇列裡面的資料
127.0.0.1:6379> xreadgroup group group1 c2 block 0  streams mq > # 開啟阻塞監聽狀態


127.0.0.1:6379> xadd mq * name nagisa age 21  # 在另一個終端中向mq中傳送一條資料
"1594970144178-0"
127.0.0.1:6379> 

127.0.0.1:6379> xreadgroup group group1 c2 block 0 streams mq >
1) 1) "mq"
   2) 1) 1) "1594970144178-0"
         2) 1) "name"
            2) "nagisa"
            3) "age"
            4) "21"
(119.02s)
127.0.0.1:6379>  # 我們看到這裡收到了資料，並且提示我們等待了119.02秒

訊息消費確認

一般訊息接收完了，我們會回覆一個確認資訊，告知我們消費完畢，命令：xack key group-key ID······

127.0.0.1:6379> xack mq group1 1594970144178-0
(integer) 1
127.0.0.1:6379>

消費確認增加了訊息的可靠性，一般在業務處理完成之後，需要執行 ack 確認訊息已經被消費完成，整個流程的執行如下圖所示：

查詢未確認的消費佇列 

127.0.0.1:6379> xpending mq group1
1) (integer) 5  # 未確認（ack）的訊息數量為 1 條
2) "1594953403230-0"
3) "1594969025661-0"
4) 1) 1) "c1"
      2) "1"
   2) 1) "c2"
      2) "3"
   3) 1) "\xe5\x8f\xa4\xe6\x98\x8e\xe5\x9c\xb0\xe8\xa7\x89"
      2) "1"
127.0.0.1:6379> 
127.0.0.1:6379> xack mq group1 1594953403230-0 1594969025661-0  # 確認兩條
(integer) 2
127.0.0.1:6379> xpending mq group1  # 還剩下三條
1) (integer) 3
2) "1594953410148-0"
3) "1594953438554-0"
4) 1) 1) "c2"
      2) "2"
   2) 1) "\xe5\x8f\xa4\xe6\x98\x8e\xe5\x9c\xb0\xe8\xa7\x89"
      2) "1"
127.0.0.1:6379>

xinfo 查詢相關命令

  • 1. 查詢流資訊：xinfo stream stream_key(佇列)
127.0.0.1:6379> xinfo stream mq
 1) "length"
 2) (integer) 6  # 佇列中有6個訊息
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1594970144178-0"
 9) "groups"  # 2個消費分組，我們上面的group1 group2
10) (integer) 2
11) "first-entry"
12) 1) "1594953403230-0"
    2) 1) "name"
       2) "satori"
       3) "age"
       4) "17"
13) "last-entry"
14) 1) "1594970144178-0"
    2) 1) "name"
       2) "nagisa"
       3) "age"
       4) "21"
127.0.0.1:6379>
  • 查詢消費組訊息：xinfo groups stream_key
127.0.0.1:6379> xinfo groups mq
1) 1) "name"
   2) "group1"  # 訊息分組名稱
   3) "consumers"
   4) (integer) 3  # 3個消費者
   5) "pending"
   6) (integer) 3  # 三個未確認的訊息
   7) "last-delivered-id"
   8) "1594970144178-0"
2) 1) "name"
   2) "group2"
   3) "consumers"
   4) (integer) 0
   5) "pending"
   6) (integer) 0
   7) "last-delivered-id"
   8) "1594953438554-0"
127.0.0.1:6379>
  • 查詢消費組成員資訊：xinfo consumers stream_key group_key
127.0.0.1:6379> xinfo consumers mq group1
1) 1) "name"
   2) "c1"  # 消費者名稱
   3) "pending"
   4) (integer) 0
   5) "idle"
   6) (integer) 25214247
2) 1) "name"
   2) "c2"
   3) "pending"
   4) (integer) 2
   5) "idle"
   6) (integer) 24613903
3) 1) "name"
   2) "\xe5\x8f\xa4\xe6\x98\x8e\xe5\x9c\xb0\xe8\xa7\x89"  # 中文名的消費者
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 25200911
127.0.0.1:6379>
  • 刪除消費者：xgroup delconsumer stream-key group-key consumer-key
127.0.0.1:6379> xgroup delconsumer mq group1 c2
  • 刪除消費組：xgroup destroy stream-key group-key
127.0.0.1:6379> xgroup destroy mq group1
(integer) 1
127.0.0.1:6379>

小結

感覺訊息佇列的話，釋出訂閱和主題訂閱不是很難，但是stream的命令有點頭疼，這裡介紹的不是很詳細。建議自己理解一下，並實際動手操作。

相關推薦

12. Redis中的訊息佇列

楔子 Redis雖然是一個快取，但是它也可以作為一個訊息佇列。所以redis還是比較有野心的，本來在快取方面就已經把memcached給幹掉了，但還想在訊息佇列的方向上闖一闖。不過雖說Redis支援訊息佇列，但是它還是作為快取

基於python實現操作redis及訊息佇列

操作 redis import redis redisPool = redis.ConnectionPool(host=\'192.168.100.50\',port=6379,db=8) redis= redis.Redis(connection_pool=redisPool)

μC/OS-II系統中訊息佇列的使用

以下內容主要注重應用，對原始碼不做分析，對原始碼有興趣的可參考官方具體文件，相關連結：https://doc.micrium.com/display/ucos/

python分散式爬蟲中訊息佇列知識點詳解

當排隊等待人數過多的時候，我們需要設定一個等待區防止秩序混亂，同時再有新來的想要排隊也可以呆在這個地方。那麼在python分散式爬蟲中，訊息佇列就相當於這樣的一個區域，爬蟲要進入這個區域找尋自己想要的資源，

TP6 基於 redis 實現訊息佇列和延遲佇列

think-queue 安裝 composer require topthink/think-queue我們需要在安裝好的think-queue > src下找到config.php複製裡面的內容，然後在根目錄下config目錄下建立queue.php檔案，將複製的內容貼上進去

Redis實現訊息佇列

今天和大家來聊一聊Redis的Stream型別，Redis從5.0開始引入了一種新的資料型別Stream型別，它是專門為訊息佇列設計的資料型別。

springboot專案使用redis進行訊息的生產和消費(redis的訊息佇列)

首先需要先把資料快取在redis裡面 工具類 /** * 快取List資料 * @param key * @param value * @param <T>

redis實現訊息佇列&釋出/訂閱模式使用 java

redis實現訊息佇列&釋出/訂閱模式使用java 　　在專案中用到了redis作為快取，再學習了ActiveMq之後想著用redis實現簡單的訊息佇列，下面做記錄。

redis中佇列訊息實現應用解耦的方法

訊息佇列的場景 我們都知道，訊息是在兩臺計算機之間傳送的資料單位，這個“訊息”可以非常簡單，例如只包含文字字串，也可以更復雜，可能包含嵌入物件。而所謂的“訊息佇列”是在訊息的傳輸過程中儲存訊息的容器。在

什麼鬼，面試官竟然讓我用Redis實現一個訊息佇列？

[ 轉載 ]原文連結：blog.csdn.net/yelvgou9995… 眾所周知，redis是一個高效能的分散式key-value儲存系統，在NoSQL資料庫市場上，redis自己就佔據了將近半壁江山，足以見到其強大之處。同時，由於redis的單執行緒特性

什麼鬼，面試官竟然讓我用Redis實現一個訊息佇列！！？

GitHub 9.4k Star 的Java工程師成神之路 ，不來瞭解一下嗎? GitHub 9.4k Star 的Java工程師成神之路 ，真的不來瞭解一下嗎?

golang實現redis的延時訊息佇列功能

前言 在學習過程中發現redis的zset還可以用來實現輕量級的延時訊息佇列功能，雖然可靠性還有待提高，但是對於一些對資料可靠性要求不那麼高的功能要求完全可以實現。本次主要採用了redis中zset中的zadd,zrangebyscor

如何處理訊息佇列消費過程中的重複訊息&如何實現冪等性

什麼是冪等 冪等本來是數學上的概念，它的定義是這樣的：如果一個函式 f(x) 滿足：f(f(x)) = f(x)，則函式 f(x) 滿足冪等性。

golang實現redis的延時訊息佇列功能示例

前言 在學習過程中發現redis的zset還可以用來實現輕量級的延時訊息佇列功能，雖然可靠性還有待提高，但是對於一些對資料可靠性要求不那麼高的功能要求完全可以實現。本次主要採用了redis中zset中的zadd,zrangebyscor

使用 Redis 流實現訊息佇列的程式碼

在介紹了 Redis 流的基本功能之後， 現在是時候使用這些功能來構建一些實際的應用了。 訊息佇列作為流的典型應用之一， 具有非常好的示範性， 因此我們將使用 Redis 流的相關功能構建一個訊息佇列應用， 這個訊息佇列

訊息佇列中，如何保證訊息的順序性？

問：如何保證訊息的順序性？ 面試官心理分析 其實這個也是用 MQ 的時候必問的話題，第一看看你了不瞭解順序這個事兒？第二看看你有沒有辦法保證訊息是有順序的？這是生產系統中常見的問題。

Redis 事務 & 訊息佇列

Redis 訊息佇列介紹 什麼是訊息佇列 訊息佇列（Message Queue）是一種應用間的通訊方式，訊息傳送後可以立即返回，有訊息系統來確保資訊的可靠傳遞，訊息生產者只管把訊息釋出到訊息佇列中而不管誰來取，訊息消費者只

redis訊息佇列

1.什麼是訊息佇列 在生活中，其實有很多的例子，都類似訊息佇列。 比如：工廠生產出來的麵包，交給超市，商場來出售，客戶通過超市，商場來買麵包，客戶不會針對某一個工廠去選擇，只管從超市買出來，工廠也不會管是

03訊息佇列系列-RabbitMQ在Spring boot中的使用

03訊息佇列系列-RabbitMQ在Spring boot中的使用 參考連結：Springboot 整合RabbitMq 一、專案簡介

Golang中優秀的訊息佇列NSQ基礎安裝及使用詳解

前言 NSQ是Go語言編寫的，開源的分散式訊息佇列中介軟體，其設計的目的是用來大規模地處理每天數以十億計級別的訊息。NSQ 具有分散式和去中心化拓撲結構，該結構具有無單點故障、故障容錯、高可用性以及能夠保證訊息