今天和大家來聊一聊Redis的Stream類型，Redis從5.0開始引入了一種新的數據類型Stream類型，它是專門為消息隊列設計的數據類型。
首先，我們先來看一下消息隊列存取消息的過程。在分散式系統中，當兩個組件要基於消息隊列進行通訊時，一個組件把消息發送到消息隊列，我們稱之為生產者，另一個組件消費消息，然後處理，我們稱之為消費者。如下圖所示：

   消息隊列主要是解決生產者和消費者處理能力不一致的問題。也經常是大廠中必不可少的中間件。 Redis在5.0之前就有一個基於發布訂閱(pub/sub)來實現消息隊列的功能，但是，它有個缺點就是當出現網路斷開、Redis宕機等，消息就會被丟棄。而Redis Stream提供了消息的持久化和主從複製功能，可以讓任何客戶端訪問任何時刻的數據，並且能記住每一個客戶端的訪問位置，還能保證消息不丟失。
首先，我們看Redis Stream提供了哪些操作命令：

• XADD 添加消息到末尾
• XTRIM 對流進行修剪，限制長度
• XLEN 獲取流包含的元素數量，即消息長度
• XDEL 刪除消息
• XRANGE 獲取消息列表，會自動過濾已經刪除的消息
• XREAD 以阻塞或非阻塞方式獲取消息列表
• XGROUP CREATE 創建消費組
• XREADGROUP 按消費者的形式來讀取消息
• XACK 用於向消息隊列確認消息處理已完成
• XPENDING 查詢每個消費組內所有消費者已讀取但尚未確認的消息

一、常用命令

1. XADD

 XADD test * invent 5 
 1629701120782-0

  表示往名稱為test的消息隊列中插入一條消息，消息的key是invent，值是5。其中test後面的「*」表示給插入的消息自動生成一個全局唯一的ID,1629701120782-0，前半部分1629701120782表示伺服器的時間，後半部分0表示插入消息在當前毫秒內的消息序號，從0開始。1629701120782-0表示1629701120782毫秒內的第一條消息。

2. XTRIM

XTRIM test maxlen 100

 表示設置test的消息隊列的長度為100，當Stream達到最大長度時，老的消息就會被刪除。由於stream內部的實現機制，精確設置一個長度上限，消耗的資源會比較大，所以我們一般採用模糊設置的方式：XTRIM test maxlen ~ 100，意思是長度可以超過100一些，可以是103、110等，由redis自行判斷什麼時候去截斷。

3. XLEN

XLEN test

   表示獲取消息隊列test的長度。

4. XDEL

XDEL test 1629701684031-0

   表示刪除消息隊列test中，ID為1629701684031-0的消息。

5. XRANGE

XRANGE test - +

   表示獲取test指定範圍內的消息，其中「-」代表最小id，「+」代表最大id。

6. XREAD

XREAD block 10000 streams test $

   表示讀取消息，其中「$」代表最新消息，「block 10000」其中10000的單位是毫秒，代表10s。表示XREAD在讀取消息時，如果沒有消息到來，XREAD將阻塞10s，然後再返回。

7. XGROUP

XGROUP CREATE test mygroup 0

   表示給消息隊列test創建一個消費組mygroup，0表示從最開始的位置讀。

8. XREADGROUP

XREADGROUP group mygroup consumer1 streams test >

   消費組mygroup中的消費者consumer1從消息隊列test中讀取所有消息，其中「>」表示從第一條尚未被消費的消息開始讀取。
需要注意的是，消息隊列中的消息一旦被消費組裡的一個消費者消費了，就不能再被該消費組內的其他消費者讀取了。使用消費組的目的是讓組內的多個消費者共同分擔讀取消息。所以，我們通常會讓每個消費者讀取部分消息，從而實現消息讀取負載在多個消費者間是均衡分布的。

9. XPENDING

XPENDING test mygroup

   為了保證消費者在發生故障重啟後，仍然可以獲取到未處理完的消息，Streams會自動使用內部隊列，來存放消費組裡每個消費者讀取的消息，直到消費者使用XACK命令，通知Streams，消息已處理完成。當消費者重啟後，可以使用XPENDING命令查看已讀取，但是沒有被確認完成的消息。

10. XACK

XACK test mygroup 1629701684031-0

   表示消費組mygroup已經確認處理了test消息隊列中id為1629701684031-0的消息。
到此為止，我們已經了解了使用Stream類型來實現消息隊列的用法了。

二、總結

   你也許會想，我們為什麼要用Redis來做消息隊列呢？要使用消息隊列，就應該使用kafka、RabbitMQ這些專門的消息隊列中間件，而Redis更適合做快取。其實，我認為，採用什麼技術和你當前所遇到的應用場景有關，如果你的消息通訊不大，對數據丟失不敏感，那麼用redis作為消息隊列不失為一個好的方法，畢竟redis相比kafka等專業的消息系統來說，更加輕量級，維護成本也低。

   今天我們就聊到這裡，更多有趣知識，請關注【程式設計師學長】，回復【資料】，可以獲得上百本技術類電子書，設計java、redis、python、數據結構和演算法等。 

   你知道的越多，你的思維也就越開闊，我們下期再見。