1. Stream簡介

Stream是redis最復雜的一個數(shù)據(jù)結構, 也是redis 5.0的一個重要更新。Redis Stream 主要用于消息隊列（MQ，Message Queue），這樣的數(shù)據(jù)結構其實很常見, 比如騰訊云的CMQ、阿里的RocketMQ、ActiveMQ、RabbitMQ以及炙手可熱的Kafka等。

Redis 本身是有一個 Redis 發(fā)布訂閱 (pub/sub) 來實現(xiàn)消息隊列的功能，但它有個缺點就是消息無法持久化，如果出現(xiàn)網(wǎng)絡斷開、Redis 宕機等，消息就會被丟棄。簡單來說發(fā)布訂閱 (pub/sub) 可以分發(fā)消息，但無法記錄歷史消息。而 Redis Stream 提供了消息的持久化和主備復制功能，可以讓任何客戶端訪問任何時刻的數(shù)據(jù)，并且能記住每一個客戶端的訪問位置，還能保證消息不丟失。

Stream主要由消息、生產者、消費者、消費組4部分組成. 這里消費組可能讓人有些困惑, 其實就是消費組里面有多個消費者, 他們互相競爭, 當一個消費了某條消息, 消息會被放入待確認隊列, 消息隊列的迭代器就會前移, 下一個同組消費者不管是誰, 都不會再次消費這個消息, 而是下一個消息。這種概念和kafka很雷同，在某些特定場景可以使用redis的stream代替kafka等消息隊列，減少系統(tǒng)復雜性，增強系統(tǒng)的穩(wěn)定性。

（1）創(chuàng)建消息隊列

/*
 * xadd用來創(chuàng)建, 每個stream有一個唯一key, *意味著讓系統(tǒng)給你返回id, id是由unix時間和從0開始下標 
 * 組成, 也就是這一毫秒的第幾個條目. 你可以自己設定, 但是要確保嚴格單調遞增. 后面就是鍵值對, 也就 
 * 是消息本身.
 */
xadd mystream * str1 hello str2 world

每個 Stream 都有唯一的名稱，它就是 Redis 的 key，在我們首次使用 xadd 指令追加消息時自動創(chuàng)建。

（2）刪除消息隊列

del mystream

（3）刪除消息隊列中某條消息

通過xdel可以刪除消息, 但是注意, 其實沒有刪除, 只是設置了標志位.

xdel mystream streamID

（4）消費一條消息

xread count 1 streams mystream streanID

（5）消費者組消費

創(chuàng)建消費者組：

xgroup create mystream myg 0 
# 這里最后是id, 0就代表從最前面開始獲取消息, 可以寫成$, 意味著獲取新消息.

消費者組內第一個消費者讀取消息：

xreadgroup group myg alice count 2 streams mystream
# 消費者alice讀取2條消息

消費者組內第二個消費者讀取消息：

xreadgroup group myg bob count 1 streams mystream 
# 消費者bob讀取1條消息

接下來我們介紹Stream的數(shù)據(jù)結構，在介紹Stream的數(shù)據(jù)結構之前，我們先來看看字典數(shù)（Trie Tree）和基數(shù)樹（Radix Tree），Redis的消息隊列Stream主要是基于基數(shù)樹來實現(xiàn)的。

2. 字典樹（Trie Tree）

在計算機科學中，字典樹（Trie Tree），又稱前綴樹，是一種有序樹，用于保存關聯(lián)數(shù)組，可以保存一些字符串->值的對應關系?；旧希１砉δ芟嗤?，都是 key-value 映射，只不過其中的鍵通常是字符串。與二叉查找樹不同，鍵不是直接保存在節(jié)點中，而是由節(jié)點在樹中的位置決定。一個節(jié)點的所有子孫都有相同的前綴，也就是這個節(jié)點對應的字符串，而根節(jié)點對應空字符串。一般情況下，不是所有的節(jié)點都有對應的值，只有葉子節(jié)點和部分內部節(jié)點所對應的鍵才有相關的值。trie tree中的鍵通常是字符串，但也可以是其它的結構。trie的算法可以很容易地修改為處理其它結構的有序序列，比如一串數(shù)字或者形狀的排列。比如，bitwise trie中的鍵是一串位元，可以用于表示整數(shù)或者內存地址。

它的原理是將每個key拆分成每個單位長度字符，然后對應到每個分支上，分支所在的節(jié)點對應為從根節(jié)點到當前節(jié)點的拼接出的key的值。Trie樹有3個基本性質：

• 根節(jié)點不包含字符，除根節(jié)點外每一個節(jié)點都只包含一個字符；
• 從根節(jié)點到某一節(jié)點，路徑上經過的字符連接起來，為該節(jié)點對應的字符串；
• 每個節(jié)點的所有子節(jié)點包含的字符都不相同；

字典樹的結構圖如下所示：

Trie 的強大之處就在于它的時間復雜度。它的插入和查詢時間復雜度都為 O(k) ，其中 k 為 key 的長度，與 Trie 中保存了多少個元素無關。Hash 表號稱是 O(1) 的，但在計算 hash 的時候就肯定會是 O(k) ，而且還有碰撞之類的問題；Trie 的缺點是空間消耗很高。

字典樹的應用場景包括：

（1）字符串檢索：事先將已知的一些字符串（字典）的有關信息保存到trie樹里，查找另外一些未知字符串是否出現(xiàn)過或者出現(xiàn)頻率。

（2）詞頻統(tǒng)計：一個文本文件，大約有一萬行，每行一個詞，要求統(tǒng)計出其中最頻繁出現(xiàn)的前10個詞，請給出思想，給出時間復雜度分析。

（3）排序：Trie樹是一棵多叉樹，只要先序遍歷整棵樹，輸出相應的字符串便是按字典序排序的結果。

（4）字符串最長公共前綴：Trie樹利用多個字符串的公共前綴來節(jié)省存儲空間，當我們把大量字符串存儲到一棵trie樹上時，我們可以快速得到某些字符串的公共前綴。

（5）字符串搜索的前綴匹配：trie樹常用于搜索提示。如當輸入一個網(wǎng)址，可以自動搜索出可能的選擇。當沒有完全匹配的搜索結果，可以返回前綴最相似的可能。

3. 基數(shù)樹（Radix Tree）

3.1 Radix Tree

Trie樹其實依然比較浪費空間，有人曾經反饋他們在實際的項目發(fā)現(xiàn)，隨著key的數(shù)量的增加，發(fā)現(xiàn)Trie樹會占用大量的內存和空間?，F(xiàn)在我們就演繹下Trie樹是如何浪費內存和空間的。比如下面的一組數(shù)據(jù)：

{
  "deck": someValue,
  "did": someValue,
  "doe": someValue,
  "dog": someValue,
  "doge": someValue,
  "dogs": someValue
}

用Trie樹的畫法把上面的key value畫出來如下：

也許你已經發(fā)現(xiàn)了一些問題。比如"deck"這一個分支，有沒有必要一直往下來拆分嗎？還是"did"，有必要d，然后i，然后d嗎?像這樣的不可分叉的單支分支，其實完全可以合并，也就是壓縮。

這樣看起來是不是要更節(jié)省一點空間呢？這只是6個單詞的樣子，數(shù)據(jù)越多，空間節(jié)省的效果越明顯。而且這樣壓縮后，不可分叉的分支高度也變矮了。我們叫這樣的Trie樹為壓縮Trie樹（Compressed Trie Tree）。壓縮Trie樹也就是Radix樹，只不過他有多個名字，有人叫壓縮Trie樹，有人叫Radix樹，它和字典樹的不同之處在于，所有只有一個子節(jié)點的中間節(jié)點都被刪除。Redis中就用到了Radix樹。

3.2 計算機對Radix Tree的處理

因為計算機可不會像人類一樣可以通過英文像上面的圖一樣來構建樹，計算機只認識0和1。所以為了真正的了解Radix樹，我們需要知道機器是怎么讀取Radix樹的。計算機對于Radix樹的處理是以bit（或二進制數(shù)字）來讀取的。一次被對比r個bit，2的r次方是radix樹的基數(shù)。這也是基數(shù)樹的這個名字的由來?，F(xiàn)在我們把上面的三個單詞變成二進制的樣子，然后一位一位的看：

dog:  01100100 01101111 01100111
doge: 01100100 01101111 01100111 01100101
dogs: 01100100 01101111 01100111 01110011

按照字符串的比對，你會發(fā)現(xiàn)dog是dogs和doge的子串。但我們現(xiàn)在比對二進制，一位一位的比對，你會發(fā)現(xiàn)dog和doge是在第二十五位的時候不一樣的。dogs和doge是在第二十八位不一樣的。按照位的比對的結果，你會發(fā)現(xiàn)doge居然是dogs二進制子串。這就是計算機的方式。

4. 基數(shù)樹（Radix Tree）的實現(xiàn)

4.1 raxNode結構定義

raxNode是radix tree的核心數(shù)據(jù)結構，其結構體如下所示：

typedef struct raxNode {    
   uint32_t iskey:1;  
   uint32_t isnull:1;  
   uint32_t iscompr:1; 
   uint32_t size:29;  
   unsigned char data[]; 
} raxNode;
typedef struct rax {
    raxNode *head;
    uint64_t numele;
    uint64_t numnodes;
} rax;

• iskey：表示這個節(jié)點是否包含key
  • 0：沒有key
  • 1：表示從頭部到其父節(jié)點的路徑完整的存儲了key，查找的時候按子節(jié)點iskey=1來判斷key是否存在
• isnull：是否有存儲value值，比如存儲元數(shù)據(jù)就只有key，沒有value值。value值也是存儲在data中
• iscompr：是否有前綴壓縮，決定了data存儲的數(shù)據(jù)結構
• size：該節(jié)點存儲的字符個數(shù)
• data：存儲子節(jié)點的信息
  • iscompr=0：非壓縮模式下，數(shù)據(jù)格式是：[header strlen=0][abc][a-ptr][b-ptr][c-ptr](value-ptr?)，有size個字符，緊跟著是size個指針，指向每個字符對應的下一個節(jié)點。size個字符之間互相沒有路徑聯(lián)系。
  • iscompr=1：壓縮模式下，數(shù)據(jù)格式是：[header strlen=3][xyz][z-ptr](value-ptr?)，只有一個指針，指向下一個節(jié)點。size個字符是壓縮字符片段

4.2 Rax Insert

以下用幾個示例來詳解rax tree插入的流程。假設 j 是遍歷已有節(jié)點的游標，i 是遍歷新增節(jié)點的游標。

（1）場景一：只插入abcd

該場景下iscompr = 1，表示使用了壓縮前綴， 因此data域只有一個z-ptr指針，z-ptr指向的葉子節(jié)點iskey = 1。節(jié)點圖為：

（2）場景二：在abcd之后插入abcdef

從abcd父節(jié)點的每個壓縮前綴字符比較，遍歷完所有abcd節(jié)點后指向了其空子節(jié)點，j = 0， i < len(abcded)。查找到abcd的空子節(jié)點，直接將ef賦值到子節(jié)點上，成為abcd的子節(jié)點。ef節(jié)點被標記為iskey=1，用來標識abcd這個key。ef節(jié)點下再創(chuàng)建一個空子節(jié)點，iskey=1來表示abcdef這個key。節(jié)點圖為：

（3）場景三：在abcd之后插入ab

ab在abcd能找到前兩位的前綴，也就是i=len(ab)，j < len(abcd)。將abcd分割成ab和cd兩個子節(jié)點，cd也是一個壓縮前綴節(jié)點，cd同時被標記為iskey=1，來表示ab這個key。cd下掛著一個空子節(jié)點，來標記abcd這個key。節(jié)點圖為：

（4）場景四：在abcd之后插入abABC

abcABC在abcd中只找到了ab這個前綴，即i < len(abcABC)，j < len(abcd)。這個步驟有點復雜，分解一下：

• step 1：將abcd從ab之后拆分，拆分成ab、c、d 三個節(jié)點。
• step 2：c節(jié)點是一個非壓縮的節(jié)點，c掛在ab子節(jié)點上。
• step 3：d節(jié)點只有一個字符，所以也是一個非壓縮節(jié)點，掛在c子節(jié)點上。
• step 4：將ABC 拆分成了A和BC， A掛在ab子節(jié)點上，和c節(jié)點屬于同一個節(jié)點，這樣A就和c同屬于父節(jié)點ab。
• step 5：將BC作為一個壓縮前綴的節(jié)點，掛在A子節(jié)點下。
• step 6：d節(jié)點和BC節(jié)點都掛一個空子節(jié)點分別標識abcd和abcABC這兩個key。

節(jié)點圖為：

（5）場景五：在abcd之后插入Aabc

abcd和Aabc沒有前綴匹配，i = 0，j = 0。

• 將abcd拆分成a、bcd兩個節(jié)點，a節(jié)點是一個非壓縮前綴節(jié)點。
• 將Aabc拆分成A、abc兩個節(jié)點，A節(jié)點也是一個非壓縮前綴節(jié)點。
• 將A節(jié)點掛在和a相同的父節(jié)點上。
• 同上，在bcd和abc這兩個節(jié)點下掛空子節(jié)點來分別表示兩個key。

節(jié)點圖為：

4.3 Rax Remove

（1）刪除

    刪除一個key的流程比較簡單，找到iskey的節(jié)點后，向上遍歷父節(jié)點刪除非iskey的節(jié)點。如果是非壓縮的父節(jié)點并且size > 1，表示還有其他非相關的路徑存在，則需要按刪除子節(jié)點的模式去處理這個父節(jié)點，主要是做memove和realloc。

（2）合并

    刪除一個key之后需要嘗試做一些合并，以收斂樹的高度。合并的條件是：

• iskey = 1的節(jié)點不能合并
• 子節(jié)點只有一個字符
• 父節(jié)點只有一個子節(jié)點（如果父節(jié)點是壓縮前綴的節(jié)點，那么只有一個子節(jié)點，滿足條件。如果父節(jié)點是非壓縮前綴的節(jié)點，那么只能有一個字符路徑才能滿足條件）

5. 消息隊列Stream的實現(xiàn)

5.1 Stream的數(shù)據(jù)結構

/* 消息ID */
typedef struct streamID {
    uint64_t ms;        /* Unix 時間（ms） */
    uint64_t seq;       /* 序列號，該毫秒下產生的第幾個消息隊列 */
} streamID;
/* 消息隊列 */
typedef struct stream {
    rax *rax;               /* 該消息隊列指向的基數(shù)樹，保存鍵值對 */
    uint64_t length;        /* 此消息隊列里面的元素個數(shù) */
    streamID last_id;       /* 上一次訪問的消息ID，如果沒有元素則為0 */
    rax *cgroups;           /* 消費者組字典: name -> streamCG */
} stream;
/* 消費者組 */
typedef struct streamCG {
    streamID last_id;       /* 上一次分發(fā)還未確認的消息 */
    rax *pel;               /* 這個消費者組中未確認的消息列表 */
    rax *consumers;         /* 該消息者組中的所有消費者 */
} streamCG;
/* 消費者組中的一個消費者 */
typedef struct streamConsumer {
    mstime_t seen_time;         /* 該消費者上次激活的時間 */
    sds name;                   /* 消費者的名字 */
    rax *pel;                   /* 該消費者待處理的未確認的消息列表 */
} streamConsumer;

 （1）消息ID（streamID）

消息ID的形式是timestampInMillis-sequence，例如1527846880572-5，它表示當前的消息在毫米時間戳1527846880572時產生，并且是該毫秒內產生的第5條消息。消息ID可以由服務器自動生成，也可以由客戶端自己指定，但是形式必須是整數(shù)-整數(shù)，而且必須是后面加入的消息的ID要大于前面的消息ID。

（2）消息內容

 一條消息內容就是一系列的key/value組成，使用基數(shù)樹進行保存。

（3）last_id 

游標，每個消費組會有個游標 last_id，任意一個消費者讀取了消息都會使游標 last_id 往前移動。

（4）pel

 消費者(Consumer)的狀態(tài)變量，作用是維護消費者的未確認的 id。 pel記錄了當前已經被客戶端讀取的消息，但是還沒有 ack (Acknowledge character：確認字符）。

5.2 Stream的API函數(shù)

stream *streamNew(void);     // 新建一個消息隊列stream
void freeStream(stream *s);  // 釋放一個消息隊列stream
unsigned long streamLength(const robj *subject);  // 返回消息隊列的消息個數(shù)
/* 把stream中從start到end的消息發(fā)送給客戶端c，如果conut不為0，則發(fā)送conut個消息，返回發(fā)出的消息個數(shù) */
size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end, size_t count, int rev, 
                            streamCG *group, streamConsumer *consumer, int flags, streamPropInfo *spi);
/* 根據(jù)迭代器獲取消息的key和value */
void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, 
                            int64_t *fieldlen, int64_t *valuelen);
int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields)；// 根據(jù)迭代器獲取消息ID
streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id); // 創(chuàng)建一個消費者組
streamCG *streamLookupCG(stream *s, sds groupname);    // 根據(jù)消費者組名查詢消費者組
/* 根據(jù)消費者名在消費者組中進行查詢 */
streamConsumer *streamLookupConsumer(streamCG *cg, sds name, int flags, int *created); 
void streamIteratorRemoveEntry(streamIterator *si, streamID *current); // 刪除消息隊列中的一個消息
/* 添加一條消息到消息隊列中 */
int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id);
/* 從消息隊列中刪除一條消息 */
int streamDeleteItem(stream *s, streamID *id);
/* 根據(jù)長度刪除消息隊列中消息的條數(shù) */
int64_t streamTrimByLength(stream *s, long long maxlen, int approx);
/* 根據(jù)最小的消息ID刪除消息隊列中消息的條數(shù) */
int64_t streamTrimByID(stream *s, streamID minid, int approx);

參考：1）https://cloud.tencent.com/developer/article/1597128

           2）http://mysql.taobao.org/monthly/2019/04/03/

到此這篇關于一文弄懂Redis Stream消息隊列的文章就介紹到這了,更多相關Redis Stream消息隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！