Redis特殊數據類型之stream怎么應用

這篇文章主要介紹“Redis特殊數據類型之stream怎么應用”，在日常操作中，相信很多人在Redis特殊數據類型之stream怎么應用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Redis特殊數據類型之stream怎么應用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

Redis Stream 是 Redis 5.0 版本新增加的數據類型，Redis 專門為消息隊列設計的數據類型。

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

• 發布訂閱模式，不能持久化也就無法可靠的保存消息，并且對于離線重連的客戶端不能讀取歷史消息的缺陷；

• List 實現消息隊列的方式不能重復消費，一個消息消費完就會被刪除，而且生產者需要自行實現全局唯一 ID。

基于以上問題，Redis 5.0 便推出了 Stream 類型也是此版本最重要的功能，用于完美地實現消息隊列，它支持消息的持久化、支持自動生成全局唯一 ID、支持 ack 確認消息的模式、支持消費組模式等，讓消息隊列更加的穩定和可靠。

常用命令

Stream 消息隊列操作命令：

• XADD : 插入消息，保證有序，可以自動生成全局唯一 ID

• XDEL : 根據消息 ID 刪除消息；

• DEL : 刪除整個 Stream；

# XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|id field value [field value ...]
127.0.0.1:6379> XADD s1 * name sid10t
"1665047636078-0"
127.0.0.1:6379> XADD s1 * name sidiot
"1665047646214-0"
# XDEL key id [id ...]
127.0.0.1:6379> XDEL s1 1665047646214-0
(integer) 1
# DEL key [key ...]
127.0.0.1:6379> DEL s1
(integer) 1

• XLEN : 查詢消息長度；

• XREAD : 用于讀取消息，可以按 ID 讀取數據；

• XRANGE : 讀取區間消息；

• XTRIM : 裁剪隊列消息個數；

# XLEN key
127.0.0.1:6379> XLEN s1
(integer) 2
# XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]
127.0.0.1:6379> XREAD streams s1 0-0
1) 1) "s1"
   2) 1) 1) "1665047636078-0"
         2) 1) "name"
            2) "sid10t"
      2) 1) "1665047646214-0"
         2) 1) "name"
            2) "sidiot"
127.0.0.1:6379> XREAD count 1 streams s1 0-0
1) 1) "s1"
   2) 1) 1) "1665047636078-0"
         2) 1) "name"
            2) "sid10t"
    # XADD 了一條消息之后的擴展
    127.0.0.1:6379> XREAD streams s1 1665047636078-0
    1) 1) "s1"
       2) 1) 1) "1665047646214-0"
             2) 1) "name"
                2) "sidiot"
          2) 1) "1665053702766-0"
             2) 1) "age"
                2) "18"
# XRANGE key start end [COUNT count]
127.0.0.1:6379> XRANGE s1 - +
1) 1) "1665047636078-0"
   2) 1) "name"
      2) "sid10t"
2) 1) "1665047646214-0"
   2) 1) "name"
      2) "sidiot"
3) 1) "1665053702766-0"
   2) 1) "age"
      2) "18"
127.0.0.1:6379> XRANGE s1 1665047636078-0 1665047646214-0
1) 1) "1665047636078-0"
   2) 1) "name"
      2) "sid10t"
2) 1) "1665047646214-0"
   2) 1) "name"
      2) "sidiot"
# XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]
127.0.0.1:6379> XLEN s1
(integer) 3
127.0.0.1:6379> XTRIM s1 maxlen 2
(integer) 1
127.0.0.1:6379> XLEN s1
(integer) 2

• XGROUP CREATE : 創建消費者組；

• XREADGROUP : 按消費組形式讀取消息；

• XPENDING 和 XACK :

XPENDING 命令可以用來查詢每個消費組內所有消費者「已讀取、但尚未確認」的消息；

XACK 命令用于向消息隊列確認消息處理已完成；

# XGROUP CREATE key groupname id|$ [MKSTREAM] [ENTRIESREAD entries_read]
# 需要注意的是，XGROUP CREATE 的 streams 必須是一個存在的 streams，否則會報錯；
127.0.0.1:6379> XGROUP CREATE myStream cGroup-top 0-0
(error) ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically.
# 0-0 從頭開始消費，$ 從尾開始消費；
127.0.0.1:6379> XADD myStream * name sid10t
"1665057823181-0"
127.0.0.1:6379> XGROUP CREATE myStream cGroup-top 0-0
OK
127.0.0.1:6379> XGROUP CREATE myStream cGroup-tail $
OK
# XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] id [id ...]
127.0.0.1:6379> XREADGROUP Group cGroup-top name count 2 STREAMS myStream >
1) 1) "myStream"
   2) 1) 1) "1665058086931-0"
         2) 1) "name"
            2) "sid10t"
      2) 1) "1665058090167-0"
         2) 1) "name"
            2) "sidiot"

應用場景

消息隊列

生產者通過 XADD 命令插入一條消息：

# * 表示讓 Redis 為插入的數據自動生成一個全局唯一的 ID
# 往名稱為 mymq 的消息隊列中插入一條消息，消息的鍵是 name，值是 sid10t
127.0.0.1:6379> XADD mymq * name sid10t
"1665058759764-0"

插入成功后會返回全局唯一的 ID："1665058759764-0"。消息的全局唯一 ID 由兩部分組成：

• 第一部分 “1665058759764” 是數據插入時，以毫秒為單位計算的當前服務器時間；

• 第二部分表示插入消息在當前毫秒內的消息序號，這是從 0 開始編號的。例如，“1665058759764-0” 就表示在 “1665058759764” 毫秒內的第 1 條消息。

消費者通過 XREAD 命令從消息隊列中讀取消息時，可以指定一個消息 ID，并從這個消息 ID 的下一條消息開始進行讀取（注意是輸入消息 ID 的下一條信息開始讀取，不是查詢輸入 ID 的消息）。

127.0.0.1:6379> XREAD STREAMS mymq 1665058759764-0
(nil)
127.0.0.1:6379> XREAD STREAMS mymq 1665058759763-0
1) 1) "mymq"
   2) 1) 1) "1665058759764-0"
         2) 1) "name"
            2) "sid10t"

如果想要實現阻塞讀（當沒有數據時，阻塞住），可以調用 XRAED 時設定 BLOCK 配置項，實現類似于 BRPOP 的阻塞讀取操作。

比如，下面這命令，設置了 BLOCK 10000 的配置項，10000 的單位是毫秒，表明 XREAD 在讀取最新消息時，如果沒有消息到來，XREAD 將阻塞 10000 毫秒（即 10 秒），然后再返回。

# 命令最后的 $ 符號表示讀取最新的消息
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS mymq $
(nil)
(10.01s)

Stream 的基礎方法，使用 xadd 存入消息和 xread 循環阻塞讀取消息的方式可以實現簡易版的消息隊列，交互流程如下圖所示：

前面介紹的這些操作 List 也支持的，接下來看看 Stream 特有的功能。

Stream 可以以使用 XGROUP 創建消費組，創建消費組之后，Stream 可以使用 XREADGROUP 命令讓消費組內的消費者讀取消息。

創建兩個消費組，這兩個消費組消費的消息隊列是 mymq，都指定從第一條消息開始讀取：

# 創建一個名為 group1 的消費組，0-0 表示從第一條消息開始讀取。
127.0.0.1:6379> XGROUP CREATE mymq group1 0-0
OK
# 創建一個名為 group2 的消費組，0-0 表示從第一條消息開始讀取。
127.0.0.1:6379> XGROUP CREATE mymq group2 0-0
OK

消費組 group1 內的消費者 consumer1 從 mymq 消息隊列中讀取所有消息的命令如下：

# 命令最后的參數“>”，表示從第一條尚未被消費的消息開始讀取。
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1665058759764-0"
         2) 1) "name"
            2) "sid10t"

消息隊列中的消息一旦被消費組里的一個消費者讀取了，就不能再被該消費組內的其他消費者讀取了，即同一個消費組里的消費者不能消費同一條消息。

比如說，我們執行完剛才的 XREADGROUP 命令后，再執行一次同樣的命令，此時讀到的就是空值了：

127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
(nil)

但是，不同消費組的消費者可以消費同一條消息（但是有前提條件，創建消息組的時候，不同消費組指定了相同位置開始讀取消息） 。

比如說，剛才 group1 消費組里的 consumer1 消費者消費了一條 id 為 1665058759764-0 的消息，現在用 group2 消費組里的 consumer1 消費者消費消息：

127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1665058759764-0"
         2) 1) "name"
            2) "sid10t"

因為我創建兩組的消費組都是從第一條消息開始讀取，所以可以看到第二組的消費者依然可以消費 id 為 1665058759764-0 的這一條消息。因此，不同的消費組的消費者可以消費同一條消息。

使用消費組的目的是讓組內的多個消費者共同分擔讀取消息，所以，我們通常會讓每個消費者讀取部分消息，從而實現消息讀取負載在多個消費者間是均衡分布的。

例如，我們執行下列命令，讓 group2 中的 consumer1、2、3 各自讀取一條消息。

# 讓 group2 中的 consumer1 從 mymq 消息隊列中消費一條消息
127.0.0.1:6379> XREADGROUP GROUP group2 consumer1 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1665060632864-0"
         2) 1) "name"
            2) "sid10t"
            
# 讓 group2 中的 consumer2 從 mymq 消息隊列中消費一條消息
127.0.0.1:6379> XREADGROUP GROUP group2 consumer2 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1665060633903-0"
         2) 1) "name"
            2) "sid10t"
            
# 讓 group2 中的 consumer3 從 mymq 消息隊列中消費一條消息
127.0.0.1:6379> XREADGROUP GROUP group2 consumer3 COUNT 1 STREAMS mymq >
1) 1) "mymq"
   2) 1) 1) "1665060634962-0"
         2) 1) "name"
            2) "sid10t"

基于 Stream 實現的消息隊列，如何保證消費者在發生故障或宕機再次重啟后，仍然可以讀取未處理完的消息？

Streams 會自動使用內部隊列（也稱為 PENDING List）留存消費組里每個消費者讀取的消息，直到消費者使用 XACK 命令通知 Streams “消息已經處理完成”。

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

如果消費者沒有成功處理消息，它就不會給 Streams 發送 XACK 命令，消息仍然會留存。此時，消費者可以在重啟后，用 XPENDING 命令查看已讀取、但尚未確認處理完成的消息。

例如，我們來查看一下 group2 中各個消費者已讀取、但尚未確認的消息個數，命令如下：

127.0.0.1:6379> XPENDING mymq group2
1) (integer) 4
2) "1665058759764-0"
3) "1665060634962-0"
4) 1) 1) "consumer1"
      2) "2"
   2) 1) "consumer2"
      2) "1"
   3) 1) "consumer3"
      2) "1"

如果想查看某個消費者具體讀取了哪些數據，可以執行下面的命令：

# 查看 group2 里 consumer2 已從 mymq 消息隊列中讀取了哪些消息
127.0.0.1:6379> XPENDING mymq group2 - + 10 consumer2
1) 1) "1665060633903-0"
   2) "consumer2"
   3) (integer) 1888805
   4) (integer) 1

可以看到，consumer2 已讀取的消息的 ID 是 1665060633903-0。

一旦消息 1665060633903-0 被 consumer2 處理了，consumer2 就可以使用 XACK 命令通知 Streams，然后這條消息就會被刪除。

127.0.0.1:6379> XACK mymq group2 1665060633903-0
(integer) 1

當我們再使用 XPENDING 命令查看時，就可以看到，consumer2 已經沒有已讀取、但尚未確認處理的消息了。

127.0.0.1:6379> XPENDING mymq group2 - + 10 consumer2
(empty array)

小結

好了，基于 Stream 實現的消息隊列就說到這里了，小結一下：

• 消息保序：XADD/XREAD

• 阻塞讀取：XREAD block

• 重復消息處理：Stream 在使用 XADD 命令，會自動生成全局唯一 ID；

• 消息可靠性：內部使用 PENDING List 自動保存消息，使用 XPENDING 命令查看消費組已經讀取但是未被確認的消息，消費者使用 XACK 確認消息；

• 支持消費組形式消費數據

Redis 基于 Stream 消息隊列與專業的消息隊列有哪些差距？

一個專業的消息隊列，必須要做到兩大塊：

• 消息不可丟。

• 消息可堆積。

1、Redis Stream 消息會丟失嗎？

使用一個消息隊列，其實就分為三大塊：生產者、隊列中間件、消費者，所以要保證消息就是保證三個環節都不能丟失數據。

Redis Stream 消息隊列能不能保證三個環節都不丟失數據？

• Redis 生產者會不會丟消息？生產者會不會丟消息，取決于生產者對于異常情況的處理是否合理。 從消息被生產出來，然后提交給 MQ 的過程中，只要能正常收到 （ MQ 中間件） 的 ack 確認響應，就表示發送成功，所以只要處理好返回值和異常，如果返回異常則進行消息重發，那么這個階段是不會出現消息丟失的。

• Redis 消費者會不會丟消息？不會，因為 Stream （ MQ 中間件）會自動使用內部隊列（也稱為 PENDING List）留存消費組里每個消費者讀取的消息，但是未被確認的消息。消費者可以在重啟后，用 XPENDING 命令查看已讀取、但尚未確認處理完成的消息。等到消費者執行完業務邏輯后，再發送消費確認 XACK 命令，也能保證消息的不丟失。

• Redis 消息中間件會不會丟消息？會，Redis 在以下 2 個場景下，都會導致數據丟失：

AOF 持久化配置為每秒寫盤，但這個寫盤過程是異步的，Redis 宕機時會存在數據丟失的可能；

主從復制也是異步的，主從切換時，也存在丟失數據的可能 (opens new window)。

可以看到，Redis 在隊列中間件環節無法保證消息不丟。像 RabbitMQ 或 Kafka 這類專業的隊列中間件，在使用時是部署一個集群，生產者在發布消息時，隊列中間件通常會寫「多個節點」，也就是有多個副本，這樣一來，即便其中一個節點掛了，也能保證集群的數據不丟失。

2、Redis Stream 消息可堆積嗎？

Redis 的數據都存儲在內存中，這就意味著一旦發生消息積壓，則會導致 Redis 的內存持續增長，如果超過機器內存上限，就會面臨被 OOM 的風險。

所以 Redis 的 Stream 提供了可以指定隊列最大長度的功能，就是為了避免這種情況發生。

當指定隊列最大長度時，隊列長度超過上限后，舊消息會被刪除，只保留固定長度的新消息。這么來看，Stream 在消息積壓時，如果指定了最大長度，還是有可能丟失消息的。

但 Kafka、RabbitMQ 專業的消息隊列它們的數據都是存儲在磁盤上，當消息積壓時，無非就是多占用一些磁盤空間。

因此，把 Redis 當作隊列來使用時，會面臨的 2 個問題：

• Redis 本身可能會丟數據；

• 面對消息擠壓，內存資源會緊張；

所以，能不能將 Redis 作為消息隊列來使用，關鍵看你的業務場景：

• 如果你的業務場景足夠簡單，對于數據丟失不敏感，而且消息積壓概率比較小的情況下，把 Redis 當作隊列是完全可以的。

• 如果你的業務有海量消息，消息積壓的概率比較大，并且不能接受數據丟失，那么還是用專業的消息隊列中間件吧。

補充：Redis 發布/訂閱機制為什么不可以作為消息隊列？

發布訂閱機制存在以下缺點，都是跟丟失數據有關：

• 發布/訂閱機制沒有基于任何數據類型實現，所以不具備「數據持久化」的能力，也就是發布/訂閱機制的相關操作，不會寫入到 RDB 和 AOF 中，當 Redis 宕機重啟，發布/訂閱機制的數據也會全部丟失。

• 發布訂閱模式是 “發后既忘” 的工作模式，如果有訂閱者離線重連之后不能消費之前的歷史消息。

• 當消費端有一定的消息積壓時，也就是生產者發送的消息，消費者消費不過來時，如果超過 32M 或者是 60s 內持續保持在 8M 以上，消費端會被強行斷開，這個參數是在配置文件中設置的，默認值是 client-output-buffer-limit pubsub 32mb 8mb 60。

所以，發布/訂閱機制只適合即使通訊的場景，比如構建哨兵集群 (opens new window)的場景采用了發布/訂閱機制。

到此，關于“Redis特殊數據類型之stream怎么應用”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！