這篇文章主要介紹“redis特殊數(shù)據(jù)類型之stream怎么應(yīng)用”，在日常操作中，相信很多人在Redis特殊數(shù)據(jù)類型之stream怎么應(yīng)用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Redis特殊數(shù)據(jù)類型之stream怎么應(yīng)用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！
目前成都創(chuàng)新互聯(lián)已為1000+的企業(yè)提供了網(wǎng)站建設(shè)、域名、虛擬空間、網(wǎng)站托管運營、企業(yè)網(wǎng)站設(shè)計、南樂網(wǎng)站維護等服務(wù)，公司將堅持客戶導向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。
Redis Stream 是 Redis 5.0 版本新增加的數(shù)據(jù)類型，Redis 專門為消息隊列設(shè)計的數(shù)據(jù)類型。
在 Redis 5.0 Stream 沒出來之前，消息隊列的實現(xiàn)方式都有著各自的缺陷，例如：
發(fā)布訂閱模式，不能持久化也就無法可靠的保存消息，并且對于離線重連的客戶端不能讀取歷史消息的缺陷；
List 實現(xiàn)消息隊列的方式不能重復消費，一個消息消費完就會被刪除，而且生產(chǎn)者需要自行實現(xiàn)全局唯一 ID。
基于以上問題，Redis 5.0 便推出了 Stream 類型也是此版本最重要的功能，用于完美地實現(xiàn)消息隊列，它支持消息的持久化、支持自動生成全局唯一 ID、支持 ack 確認消息的模式、支持消費組模式等，讓消息隊列更加的穩(wěn)定和可靠。
Stream 消息隊列操作命令：
XADD : 插入消息，保證有序，可以自動生成全局唯一 ID
XDEL : 根據(jù)消息 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 讀取數(shù)據(jù)；
XRANGE : 讀取區(qū)間消息；
XTRIM : 裁剪隊列消息個數(shù)；
# 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 : 創(chuàng)建消費者組；
XREADGROUP : 按消費組形式讀取消息；
XPENDING 和 XACK :
XPENDING 命令可以用來查詢每個消費組內(nèi)所有消費者「已讀取、但尚未確認」的消息；
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"
消息隊列
生產(chǎn)者通過 XADD 命令插入一條消息：
# * 表示讓 Redis 為插入的數(shù)據(jù)自動生成一個全局唯一的 ID
# 往名稱為 mymq 的消息隊列中插入一條消息，消息的鍵是 name，值是 sid10t
127.0.0.1:6379> XADD mymq * name sid10t
"1665058759764-0"
插入成功后會返回全局唯一的 ID："1665058759764-0"。消息的全局唯一 ID 由兩部分組成：
第一部分 “1665058759764” 是數(shù)據(jù)插入時，以毫秒為單位計算的當前服務(wù)器時間；
第二部分表示插入消息在當前毫秒內(nèi)的消息序號，這是從 0 開始編號的。例如，“1665058759764-0” 就表示在 “1665058759764” 毫秒內(nèi)的第 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"
如果想要實現(xiàn)阻塞讀（當沒有數(shù)據(jù)時，阻塞?。梢哉{(diào)用 XRAED 時設(shè)定 BLOCK 配置項，實現(xiàn)類似于 BRPOP 的阻塞讀取操作。
比如，下面這命令，設(shè)置了 BLOCK 10000 的配置項，10000 的單位是毫秒，表明 XREAD 在讀取最新消息時，如果沒有消息到來，XREAD 將阻塞 10000 毫秒（即 10 秒），然后再返回。
# 命令最后的 $ 符號表示讀取最新的消息
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS mymq $
(nil)
(10.01s)
Stream 的基礎(chǔ)方法，使用 xadd 存入消息和 xread 循環(huán)阻塞讀取消息的方式可以實現(xiàn)簡易版的消息隊列，交互流程如下圖所示：
前面介紹的這些操作 List 也支持的，接下來看看 Stream 特有的功能。
Stream 可以以使用 XGROUP 創(chuàng)建消費組，創(chuàng)建消費組之后，Stream 可以使用 XREADGROUP 命令讓消費組內(nèi)的消費者讀取消息。
創(chuàng)建兩個消費組，這兩個消費組消費的消息隊列是 mymq，都指定從第一條消息開始讀?。?/p>
# 創(chuàng)建一個名為 group1 的消費組，0-0 表示從第一條消息開始讀取。
127.0.0.1:6379> XGROUP CREATE mymq group1 0-0
OK
# 創(chuàng)建一個名為 group2 的消費組，0-0 表示從第一條消息開始讀取。
127.0.0.1:6379> XGROUP CREATE mymq group2 0-0
OK
消費組 group1 內(nèi)的消費者 consumer1 從 mymq 消息隊列中讀取所有消息的命令如下：
# 命令最后的參數(shù)“>”，表示從第一條尚未被消費的消息開始讀取。
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
1) 1) "mymq"
2) 1) 1) "1665058759764-0"
2) 1) "name"
2) "sid10t"
消息隊列中的消息一旦被消費組里的一個消費者讀取了，就不能再被該消費組內(nèi)的其他消費者讀取了，即同一個消費組里的消費者不能消費同一條消息。
比如說，我們執(zhí)行完剛才的 XREADGROUP 命令后，再執(zhí)行一次同樣的命令，此時讀到的就是空值了：
127.0.0.1:6379> XREADGROUP GROUP group1 consumer1 STREAMS mymq >
(nil)
但是，不同消費組的消費者可以消費同一條消息（但是有前提條件，創(chuàng)建消息組的時候，不同消費組指定了相同位置開始讀取消息） 。
比如說，剛才 group1 消費組里的 consumer1 消費者消費了一條 id 為 1665058759764-0 的消息，現(xiàn)在用 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"
因為我創(chuàng)建兩組的消費組都是從第一條消息開始讀取，所以可以看到第二組的消費者依然可以消費 id 為 1665058759764-0 的這一條消息。因此，不同的消費組的消費者可以消費同一條消息。
使用消費組的目的是讓組內(nèi)的多個消費者共同分擔讀取消息，所以，我們通常會讓每個消費者讀取部分消息，從而實現(xiàn)消息讀取負載在多個消費者間是均衡分布的。
例如，我們執(zhí)行下列命令，讓 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 實現(xiàn)的消息隊列，如何保證消費者在發(fā)生故障或宕機再次重啟后，仍然可以讀取未處理完的消息？
Streams 會自動使用內(nèi)部隊列（也稱為 PENDING List）留存消費組里每個消費者讀取的消息，直到消費者使用 XACK 命令通知 Streams “消息已經(jīng)處理完成”。
消費確認增加了消息的可靠性，一般在業(yè)務(wù)處理完成之后，需要執(zhí)行 XACK 命令確認消息已經(jīng)被消費完成，整個流程的執(zhí)行如下圖所示：
如果消費者沒有成功處理消息，它就不會給 Streams 發(fā)送 XACK 命令，消息仍然會留存。此時，消費者可以在重啟后，用 XPENDING 命令查看已讀取、但尚未確認處理完成的消息。
例如，我們來查看一下 group2 中各個消費者已讀取、但尚未確認的消息個數(shù)，命令如下：
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"
如果想查看某個消費者具體讀取了哪些數(shù)據(jù)，可以執(zhí)行下面的命令：
# 查看 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 已經(jīng)沒有已讀取、但尚未確認處理的消息了。
127.0.0.1:6379> XPENDING mymq group2 - + 10 consumer2
(empty array)
好了，基于 Stream 實現(xiàn)的消息隊列就說到這里了，小結(jié)一下：
消息保序：XADD/XREAD
阻塞讀取：XREAD block
重復消息處理：Stream 在使用 XADD 命令，會自動生成全局唯一 ID；
消息可靠性：內(nèi)部使用 PENDING List 自動保存消息，使用 XPENDING 命令查看消費組已經(jīng)讀取但是未被確認的消息，消費者使用 XACK 確認消息；
支持消費組形式消費數(shù)據(jù)
Redis 基于 Stream 消息隊列與專業(yè)的消息隊列有哪些差距？
一個專業(yè)的消息隊列，必須要做到兩大塊：
消息不可丟。
消息可堆積。
1、Redis Stream 消息會丟失嗎？
使用一個消息隊列，其實就分為三大塊：生產(chǎn)者、隊列中間件、消費者，所以要保證消息就是保證三個環(huán)節(jié)都不能丟失數(shù)據(jù)。
Redis Stream 消息隊列能不能保證三個環(huán)節(jié)都不丟失數(shù)據(jù)？
Redis 生產(chǎn)者會不會丟消息？生產(chǎn)者會不會丟消息，取決于生產(chǎn)者對于異常情況的處理是否合理。 從消息被生產(chǎn)出來，然后提交給 MQ 的過程中，只要能正常收到 （ MQ 中間件） 的 ack 確認響應(yīng)，就表示發(fā)送成功，所以只要處理好返回值和異常，如果返回異常則進行消息重發(fā)，那么這個階段是不會出現(xiàn)消息丟失的。
Redis 消費者會不會丟消息？不會，因為 Stream （ MQ 中間件）會自動使用內(nèi)部隊列（也稱為 PENDING List）留存消費組里每個消費者讀取的消息，但是未被確認的消息。消費者可以在重啟后，用 XPENDING 命令查看已讀取、但尚未確認處理完成的消息。等到消費者執(zhí)行完業(yè)務(wù)邏輯后，再發(fā)送消費確認 XACK 命令，也能保證消息的不丟失。
Redis 消息中間件會不會丟消息？會，Redis 在以下 2 個場景下，都會導致數(shù)據(jù)丟失：
AOF 持久化配置為每秒寫盤，但這個寫盤過程是異步的，Redis 宕機時會存在數(shù)據(jù)丟失的可能；
主從復制也是異步的，主從切換時，也存在丟失數(shù)據(jù)的可能 (opens new window)。
可以看到，Redis 在隊列中間件環(huán)節(jié)無法保證消息不丟。像 RabbitMQ 或 Kafka 這類專業(yè)的隊列中間件，在使用時是部署一個集群，生產(chǎn)者在發(fā)布消息時，隊列中間件通常會寫「多個節(jié)點」，也就是有多個副本，這樣一來，即便其中一個節(jié)點掛了，也能保證集群的數(shù)據(jù)不丟失。
2、Redis Stream 消息可堆積嗎？
Redis 的數(shù)據(jù)都存儲在內(nèi)存中，這就意味著一旦發(fā)生消息積壓，則會導致 Redis 的內(nèi)存持續(xù)增長，如果超過機器內(nèi)存上限，就會面臨被 OOM 的風險。
所以 Redis 的 Stream 提供了可以指定隊列最大長度的功能，就是為了避免這種情況發(fā)生。
當指定隊列最大長度時，隊列長度超過上限后，舊消息會被刪除，只保留固定長度的新消息。這么來看，Stream 在消息積壓時，如果指定了最大長度，還是有可能丟失消息的。
但 Kafka、RabbitMQ 專業(yè)的消息隊列它們的數(shù)據(jù)都是存儲在磁盤上，當消息積壓時，無非就是多占用一些磁盤空間。
因此，把 Redis 當作隊列來使用時，會面臨的 2 個問題：
Redis 本身可能會丟數(shù)據(jù)；
面對消息擠壓，內(nèi)存資源會緊張；
所以，能不能將 Redis 作為消息隊列來使用，關(guān)鍵看你的業(yè)務(wù)場景：
如果你的業(yè)務(wù)場景足夠簡單，對于數(shù)據(jù)丟失不敏感，而且消息積壓概率比較小的情況下，把 Redis 當作隊列是完全可以的。
如果你的業(yè)務(wù)有海量消息，消息積壓的概率比較大，并且不能接受數(shù)據(jù)丟失，那么還是用專業(yè)的消息隊列中間件吧。
補充：Redis 發(fā)布/訂閱機制為什么不可以作為消息隊列？
發(fā)布訂閱機制存在以下缺點，都是跟丟失數(shù)據(jù)有關(guān)：
發(fā)布/訂閱機制沒有基于任何數(shù)據(jù)類型實現(xiàn)，所以不具備「數(shù)據(jù)持久化」的能力，也就是發(fā)布/訂閱機制的相關(guān)操作，不會寫入到 RDB 和 AOF 中，當 Redis 宕機重啟，發(fā)布/訂閱機制的數(shù)據(jù)也會全部丟失。
發(fā)布訂閱模式是 “發(fā)后既忘” 的工作模式，如果有訂閱者離線重連之后不能消費之前的歷史消息。
當消費端有一定的消息積壓時，也就是生產(chǎn)者發(fā)送的消息，消費者消費不過來時，如果超過 32M 或者是 60s 內(nèi)持續(xù)保持在 8M 以上，消費端會被強行斷開，這個參數(shù)是在配置文件中設(shè)置的，默認值是 client-output-buffer-limit pubsub 32mb 8mb 60。
所以，發(fā)布/訂閱機制只適合即使通訊的場景，比如構(gòu)建哨兵集群 (opens new window)的場景采用了發(fā)布/訂閱機制。
到此，關(guān)于“Redis特殊數(shù)據(jù)類型之stream怎么應(yīng)用”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
本文標題：Redis特殊數(shù)據(jù)類型之stream怎么應(yīng)用
地址分享：http://weahome.cn/article/gehhhe.html
重慶分公司
重慶分公司
