在redis5.0之前，如果想使用它作為簡(jiǎn)單的消息隊(duì)列，最好的選擇就是自身提供的pub/sub模式.它支持簡(jiǎn)單的發(fā)布/訂閱模式，發(fā)布一個(gè)channel綁定一條消息，然后可以有多個(gè)消費(fèi)者監(jiān)聽這個(gè)channel，每個(gè)消費(fèi)者都能收到相同的消息。不支持持久化，不支持查詢,不支持分組，不支持分片消費(fèi)，也沒(méi)有提供很好的監(jiān)控手段（有簡(jiǎn)單的pubsub容器命令，可以看有哪些channel，訂閱者數(shù)量等）。但是5.0之后，倘若我們?nèi)巳赃x擇redis作為簡(jiǎn)單消息隊(duì)列，就可以使用新的數(shù)據(jù)類型STREAM

STREAM數(shù)據(jù)類型介紹

數(shù)據(jù)類型基礎(chǔ)說(shuō)明

• 可以理解為一個(gè)有時(shí)間序列的一組數(shù)據(jù)集合，每一條新增的數(shù)據(jù)都是追加到數(shù)據(jù)集末尾，每一條數(shù)據(jù)都有自己的唯一id
• 底層數(shù)據(jù)結(jié)構(gòu)是基數(shù)樹
• 一個(gè)Stream可以有多個(gè)消費(fèi)者分組group，每一個(gè)group也可以有多個(gè)消費(fèi)者consumer，支持分片讀取，全部讀取，按照ID分段讀取
• 隨機(jī)訪問(wèn)時(shí)間復(fù)雜度是O(1),向流中添加一個(gè)條目的時(shí)間為O(1)。 訪問(wèn)任意一項(xiàng)的時(shí)間為O(n)，其中n是ID的長(zhǎng)度.

常用命令及詳解

• XADD 向指定的 Stream 添加一條新消息。
  XADD key [MAXLEN [~] count] * field1 value1 [field2 value2 ...]參數(shù)說(shuō)明：
  key：Stream 的名稱。
  MAXLEN [~] count：可選，限制 Stream 最大長(zhǎng)度，超出自動(dòng)裁剪最老消息。~ 表示近似修剪，性能更優(yōu)?！緦?shí)際上使用要注意，超過(guò)最大值直接丟棄，也就是“消失了“】
  *：讓 Redis 自動(dòng)生成消息ID，也可自定義ID。
  field value：消息體的鍵值對(duì)。
  用法舉例：XADD mystream * name Alice age 20

• XRANGE 按ID范圍讀取 Stream 中的消息
  XRANGE key start end [COUNT count]參數(shù)說(shuō)明：
  start、end：起止ID，- 表示最小ID，+ 表示最大ID。
  COUNT：可選，限制返回條數(shù)。
  用法舉例：XRANGE mystream - + # 讀取所有消息

• XREAD 從一個(gè)或多個(gè) Stream 讀取新消息，可阻塞等待
  XREAD [BLOCK milliseconds] STREAMS key [key ...] id [id ...]參數(shù)說(shuō)明：
  BLOCK：可選，阻塞等待新消息的毫秒數(shù)。
  STREAMS：后面跟 Stream 名稱和起始ID。
  用法舉例：XREAD BLOCK 5000 STREAMS mystream $ $ 表示只讀新消息

• XGROUP 創(chuàng)建、刪除、管理 Stream 的消費(fèi)者組。
  XGROUP CREATE mystream mygroup 0-0 MKSTREAM常用子命令：

  • 創(chuàng)建組：XGROUP CREATE mystream mygroup 0-0 MKSTREAM0-0：從頭消費(fèi)；$：只消費(fèi)新消息。
    MKSTREAM：Stream 不存在時(shí)自動(dòng)創(chuàng)建
  • 刪除組：XGROUP DESTROY mystream mygroup
  • 創(chuàng)建消費(fèi)者、刪除消費(fèi)者。一般不需要，會(huì)自動(dòng)創(chuàng)建
    XGROUP CREATECONSUMER mystream mygroup consumer-1XGROUP CREATECONSUMER mystream mygroup consumer-1

• XREADGROUP 以消費(fèi)者組身份讀取消息，實(shí)現(xiàn)分布式并發(fā)消費(fèi)
  XREADGROUP GROUP group consumer [BLOCK milliseconds] STREAMS key [key ...] id [id ...]參數(shù)說(shuō)明：
  GROUP group consumer：指定組名和消費(fèi)者名。
  id：> 表示只讀未分配的新消息，其他ID（如0）可用于補(bǔ)償pending。
  舉例：XREADGROUP GROUP mygroup consumer-1 BLOCK 5000 STREAMS mystream >

• XPENDING 查看某個(gè)組下所有未ack的消息（即已分配但未確認(rèn)）注意這里不是消息的快照，它只是存儲(chǔ)消息的ID列表，并不會(huì)復(fù)制一份消息內(nèi)容XPENDING key group [start end count [consumer]]舉例：XPENDING mystream mygroup - + 10
  XPENDING mystream mygroup - + 10 consumer-1

• XACK 用于確認(rèn)消息已被消費(fèi)，也就是從pending狀態(tài)PEL中移除
  舉例：XACK mystream mygroup 1680000000000-0

• XCLAIM/XAUTOCLAIM 將長(zhǎng)時(shí)間未ack的pending消息轉(zhuǎn)移到其他消費(fèi)者/實(shí)現(xiàn)自動(dòng)補(bǔ)償。
  舉例：XCLAIM mystream mygroup consumer-2 60000 1680000000000-0XAUTOCLAIM mystream mygroup consumer-2 60000 0-0 COUNT 10

• XTRIM 限制流的最大長(zhǎng)度，自動(dòng)刪除最老的消息。無(wú)論是否被ack的消息，都會(huì)被裁減。
  語(yǔ)法：XTRIM key MAXLEN [~] count
  舉例：XTRIM mystream MAXLEN ~ 1000

• XDEL 從Stream中刪除指定ID的消息，可以一次刪除多個(gè),用空格隔開即可
  XDEL mystream 1680000000000-0

實(shí)際使用場(chǎng)景

可用作消息隊(duì)列

• 當(dāng)需要一個(gè)輕量級(jí)的、安全性要求比較低、可靠性不要求那么高的一個(gè)消息隊(duì)列時(shí)，使用stream就很合適，性能也非常不錯(cuò)，單機(jī)能支持每秒幾十萬(wàn)的寫入
• 典型場(chǎng)景：訂單異步處理、短信/郵件通知、日志收集、任務(wù)分發(fā)等

可以作為事件總線

• 作為事件總線，支撐微服務(wù)間的事件發(fā)布與訂閱，作為事件源（例如，跟蹤用戶操作、點(diǎn)擊等）。
• 例如：用戶注冊(cè)事件、支付完成事件等，多個(gè)服務(wù)可并發(fā)消費(fèi)

延遲隊(duì)列/死信隊(duì)列

• 利用 Stream 的 pending/ack/xclaim 機(jī)制實(shí)現(xiàn)可靠的延遲消息、死信消息補(bǔ)償。

實(shí)時(shí)數(shù)據(jù)流處理

• IoT、監(jiān)控、風(fēng)控等場(chǎng)景下，設(shè)備/傳感器數(shù)據(jù)實(shí)時(shí)寫入 Stream，后端實(shí)時(shí)消費(fèi)分析。
• 支持高并發(fā)寫入和多消費(fèi)者并發(fā)處理

重要說(shuō)明

關(guān)于持久化和消息刪除

• 消息是默認(rèn)就持久化的，并且并不提供設(shè)置過(guò)期時(shí)間，那么如果在消息量大且請(qǐng)求量大的情況下，會(huì)占用很多內(nèi)存
• 如果在新增消息的時(shí)候使用maxlen選項(xiàng)限定了stream的長(zhǎng)度，那么一定要考慮使用多個(gè)consumer，而且要提供一定的處理機(jī)制在某些consumer不可用的時(shí)候，將消息XCLAIM到可用的消費(fèi)者。避免超過(guò)限定長(zhǎng)度后，丟失消息。
• 不推薦每次消費(fèi)完成后使用Xdel去刪除，而是采用Xtrim收縮，結(jié)合Xinfo、Xlen等命令定期檢測(cè)stream的長(zhǎng)度，然后根據(jù)實(shí)際情況設(shè)置合理的收縮長(zhǎng)度，定期的清理不再使用的消息。因?yàn)榧词故褂肵del取刪除消息，在當(dāng)前的實(shí)現(xiàn)中，直到宏節(jié)點(diǎn)完全為空時(shí)才真正回收內(nèi)存

讀取的阻塞和非阻塞

• XRANGE 、XREAD 或 XREADGROUP ，沒(méi)有BLOCK選項(xiàng)時(shí)，像任何其他Redis命令一樣同步調(diào)用，此時(shí)他們就是同步命令；如果加上BLOCK選項(xiàng)就時(shí)非阻塞的，等待指定的毫秒直到有可以消費(fèi)的消息并立即返回

插入的性能

• XADD 非?？欤绻褂昧魉€，在普通機(jī)器中每秒可以輕松插入50萬(wàn)到100萬(wàn)項(xiàng)
• 以下是官網(wǎng)提供的延遲測(cè)試結(jié)果：【在這里，我們每次迭代最多處理10k條消息，這意味著 XREADGROUP 的 COUNT 參數(shù)被設(shè)置為10000。這增加了大量的延遲，但為了讓緩慢的消費(fèi)者能夠跟上消息流，這是必需的。因此，你可以預(yù)期真實(shí)世界的延遲要小得多】

Results obtained: 結(jié)果:
		Processed between 0 and 1 ms -> 74.11%
		Processed between 1 and 2 ms -> 25.80%
		Processed between 2 and 3 ms -> 0.06%
		Processed between 3 and 4 ms -> 0.01%
		Processed between 4 and 5 ms -> 0.02%
		因此，99.9%的請(qǐng)求的延遲<= 2毫秒，異常值仍然非常接近平均值。

• 另外需要注意的是，從Redis 6.2.0版本開始，才增加了 IDLE 選項(xiàng)和獨(dú)占范圍間隔，雖然5.0就引入了stream數(shù)據(jù)類型

消費(fèi)者組

• 何時(shí)不需要消費(fèi)者組：如果你有一個(gè)數(shù)據(jù)流和多個(gè)客戶端，而且你希望所有客戶端都能收到所有信息，那么你就不需要消費(fèi)者組。
• 如果你有一個(gè)數(shù)據(jù)流和多個(gè)客戶端，而且你希望在客戶端之間對(duì)數(shù)據(jù)流進(jìn)行分區(qū)或分片，以便每個(gè)客戶端都能獲得到達(dá)數(shù)據(jù)流的消息的子集，那么你就需要一個(gè)消費(fèi)者組。
• 當(dāng)使用 XREADGROUP 讀取時(shí)，服務(wù)器將記錄哪些消息給到了哪些消費(fèi)者：消息將存儲(chǔ)在使用者組內(nèi)的 Pending Entries List （PEL） 中，該列表是已傳遞但尚未確認(rèn)的消息 ID 列表。
• 當(dāng)實(shí)際場(chǎng)景是：可靠性不是必需的，并且偶爾的消息丟失是可以接受的情況下，可以使用 NOACK 子命令來(lái)避免將消息添加到 PEL。這相當(dāng)于在讀取消息時(shí)確認(rèn)消息（自動(dòng)ACK）。
• 使用 XREADGROUP 時(shí)，在 STREAMS 選項(xiàng)中指定的 ID 可以是以下兩種之一：

  特殊的 > ID，表示消費(fèi)者只想接收從未發(fā)送給其他消費(fèi)者的信息。它的意思是，給我新郵件。

  任何其他 ID，即 0 或任何其他有效 ID 或不完整 ID（僅毫秒時(shí)間部分），都將導(dǎo)致返回發(fā)送命令的用戶的待處理?xiàng)l目，且 ID 大于所提供的 ID。因此，基本上如果 ID 不大于，那么命令將只允許客戶訪問(wèn)其待處理?xiàng)l目：已向其發(fā)送但尚未確認(rèn)的信息。請(qǐng)注意，在這種情況下，BLOCK 和 NOACK 都會(huì)被忽略。

屬于PEL中的消息可以刪除嗎

pending狀態(tài)的消息是可以被刪除的，redis并沒(méi)有設(shè)計(jì)未確認(rèn)的消息不允許刪除。如果采用xdel刪除消息后，pending列表將仍然保留待消費(fèi)消息的ID，但是消息內(nèi)容沒(méi)有了。因此，在讀取此類PEL條目時(shí)，Redis會(huì)返回一個(gè)空值。

一個(gè)stream的一個(gè)group多個(gè)consumer時(shí)如何消費(fèi)的

1. 分區(qū)/競(jìng)爭(zhēng)消費(fèi)（Work Queue 模式）

• 每條消息只會(huì)被 group 下的一個(gè)消費(fèi)者消費(fèi)，不會(huì)被所有消費(fèi)者都消費(fèi)。
• Redis 會(huì)將新消息分配給 group 內(nèi)“空閑”的消費(fèi)者，實(shí)現(xiàn)消息的負(fù)載均衡（輪詢或空閑優(yōu)先，具體是由實(shí)現(xiàn)的客戶端決定）。
• 多個(gè)消費(fèi)者并發(fā)時(shí)，消息會(huì)被“分?jǐn)?rdquo;到各個(gè)消費(fèi)者，每個(gè)消息只會(huì)被其中一個(gè)消費(fèi)。
• 消息被轉(zhuǎn)XCLAIM到另一個(gè)消費(fèi)者時(shí)會(huì)增加投遞次數(shù)，并發(fā)時(shí)投遞次數(shù)、時(shí)間戳都會(huì)變化，因此也只有一個(gè)消費(fèi)者成功獲取。XPENDING命令就可以看到每個(gè)消息被投遞的次數(shù)

2. pending 機(jī)制

• 消費(fèi)者用 XREADGROUP 拉取消息后，消息會(huì)進(jìn)入該消費(fèi)者的 pending（未確認(rèn)）列表，直到被 XACK。
• 如果某個(gè)消費(fèi)者掛掉，pending 里的消息可以被其他消費(fèi)者用 XCLAIM/XAUTOCLAIM 方式“搶救”回來(lái)，保證消息最終被消費(fèi)。

3. 分布式環(huán)境下的存儲(chǔ)

• stream的增加數(shù)據(jù)和其他數(shù)據(jù)類型一樣，都是需要一個(gè)唯一的key，然后給key綁定指定數(shù)據(jù)類型的一個(gè)或者多個(gè)值
• 那也就是說(shuō)，即使在分布式存儲(chǔ)環(huán)境下，它和其他的key一樣，相同的key的數(shù)據(jù)一定存在同一個(gè)分片上（因?yàn)閞edis的分片機(jī)制就是按照Key來(lái)實(shí)現(xiàn)的）
• 實(shí)際使用時(shí)key的設(shè)置就要相對(duì)分散，否則數(shù)據(jù)會(huì)傾斜到某些節(jié)點(diǎn)上

x. 如果要“廣播”效果（每個(gè)消費(fèi)者都收到同一條消息），需要每個(gè)消費(fèi)者用不同的 group?；蛘叨紡V播了，就使用PUB/SUB吧，，～～

觀測(cè)流

• Redis流和消費(fèi)者組有不同的方式來(lái)觀察正在發(fā)生的事情，比如前面說(shuō)的XPENDING ，它允許我們檢查在給定時(shí)刻正在處理的消息列表，以及它們的空閑時(shí)間和交付數(shù)量
• XINFO：這個(gè)命令使用子命令來(lái)顯示流及其消費(fèi)者組狀態(tài)的不同信息。例如，XINFO流報(bào)告有關(guān)流本身的信息?？梢杂糜赟tream、Group、CONSUMERS
• 實(shí)際項(xiàng)目中可結(jié)合其他命令，直觀的展示流的各種信息，比如有多少個(gè)分組、有哪些分組、有哪些消費(fèi)者、消費(fèi)者狀態(tài)、消費(fèi)進(jìn)度、總條目數(shù)據(jù)等。有了這些信息就可以對(duì)消息的可靠性進(jìn)行分析，還能及時(shí)發(fā)現(xiàn)資源占用情況，結(jié)合定時(shí)任務(wù)等作出具體性能調(diào)整。

更加詳細(xì)stream的細(xì)節(jié)介紹，可以參考官網(wǎng)：https://redis.io/docs/latest/develop/data-types/streams

稍后我將具體介紹如何在代碼中使用stream來(lái)作為消息隊(duì)列。

到此這篇關(guān)于使用redis的stream數(shù)據(jù)類型做消息隊(duì)列的文章就介紹到這了,更多相關(guān)redis消息隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論