一、概述

消息隊(duì)列路幸，Message Queue荐开，常用于解決并發(fā)系統(tǒng)中的資源一致性問(wèn)題，提升峰值的處理能力简肴，同時(shí)保證消息的順序性晃听、可恢復(fù)性、必送達(dá)性砰识，對(duì)應(yīng)用進(jìn)行解耦能扒，或者實(shí)現(xiàn)異步通訊等。市面上的 MQ應(yīng)用有很多（例如：Kafka辫狼，RabbitMQ初斑，Disque），同時(shí)也可以基于 Redis 來(lái)實(shí)現(xiàn)膨处，比較典型的方案有：

• 基于List的 LPUSH+BRPOP 的實(shí)現(xiàn)
• PUB/SUB见秤，訂閱/發(fā)布模式
• 基于Sorted-Set的實(shí)現(xiàn)

• 基于Stream類型的實(shí)現(xiàn)
  在消息隊(duì)列使用中，有生產(chǎn)者producter和消費(fèi)者consumer真椿。生產(chǎn)者負(fù)責(zé)生成消息鹃答，消費(fèi)者負(fù)責(zé)使用處理消息。生產(chǎn)突硝，指的是將消息放入消息隊(duì)列测摔。 消費(fèi)，指的是讀取并處理消息狞换。通常一個(gè)消息再被消費(fèi)后避咆，就應(yīng)該從消息隊(duì)列中刪除。

  
  

?

二修噪、實(shí)現(xiàn)

1查库、基于List的LPUSH+BRPOP的實(shí)現(xiàn)

LPUSH，將消息放入消息隊(duì)列（生產(chǎn)者）
BRPOP黄琼，從隊(duì)列中取出消息樊销，阻塞模式（消費(fèi)者）

TBase中不支持BRPOP整慎，只支持RPOP，BRPOP是RPOP的阻塞版本
該模式優(yōu)點(diǎn)：

• 實(shí)現(xiàn)簡(jiǎn)單
• Reids支持持久化消息围苫，意味著消息不會(huì)丟失裤园，可以重復(fù)查看（注意不是消費(fèi)，只看不用剂府，LRANGE類的指令）拧揽。
• 可以保證順序，保證使用LPUSH命令腺占，可以保證消息的順序性
• 使用RPUSH淤袜，可以將消息放在隊(duì)列的開(kāi)頭，達(dá)到優(yōu)先消息的目的衰伯，可以實(shí)現(xiàn)簡(jiǎn)易的消息優(yōu)先隊(duì)列铡羡。
  該模式缺點(diǎn)：
• 做消費(fèi)確認(rèn)ACK比較麻煩，就是不能保證消費(fèi)者在讀取之后意鲸，未處理后的宕機(jī)問(wèn)題烦周。導(dǎo)致消息意外丟失。通常需要自己維護(hù)一個(gè)Pending列表怎顾，保證消息的處理確認(rèn)读慎。
• 不能做廣播模式，例如典型的Pub/Discribe模式杆勇。
• 不能重復(fù)消費(fèi)贪壳，一旦消費(fèi)就會(huì)被刪除
• 不支持分組消費(fèi)饱亿，需要自己在業(yè)務(wù)邏輯層解決
  ?

2蚜退、PUB/SUB，訂閱/發(fā)布模式

SUBSCRIBE彪笼，用于訂閱信道
PUBLISH钻注，向信道發(fā)送消息
UNSUBSCRIBE，取消訂閱

生產(chǎn)者和消費(fèi)者通過(guò)相同的一個(gè)信道（Channel）進(jìn)行交互配猫。信道其實(shí)也就是隊(duì)列幅恋。通常會(huì)有多個(gè)消費(fèi)者。多個(gè)消費(fèi)者訂閱同一個(gè)信道泵肄，當(dāng)生產(chǎn)者向信道發(fā)布消息時(shí)捆交，該信道會(huì)立即將消息逐一發(fā)布給每個(gè)消費(fèi)者「玻可見(jiàn)品追，該信道對(duì)于消費(fèi)者是發(fā)散的信道，每個(gè)消費(fèi)者都可以得到相同的消息冯丙。典型的對(duì)多的關(guān)系肉瓦。
該模式優(yōu)點(diǎn)：

• 典型的廣播模式，一個(gè)消息可以發(fā)布到多個(gè)消費(fèi)者
• 多信道訂閱，消費(fèi)者可以同時(shí)訂閱多個(gè)信道泞莉，從而接收多類消息
• 消息即時(shí)發(fā)送哪雕，消息不用等待消費(fèi)者讀取，消費(fèi)者會(huì)自動(dòng)接收到信道發(fā)布的消息
  該模式缺點(diǎn)：
• 消息一旦發(fā)布鲫趁，不能接收斯嚎。換句話就是發(fā)布時(shí)若客戶端不在線，則消息丟失挨厚，不能尋回
• 不能保證每個(gè)消費(fèi)者接收的時(shí)間是一致的
• 若消費(fèi)者客戶端出現(xiàn)消息積壓孝扛，到一定程度，會(huì)被強(qiáng)制斷開(kāi)幽崩，導(dǎo)致消息意外丟失苦始。通常發(fā)生在消息的生產(chǎn)遠(yuǎn)大于消費(fèi)速度時(shí)
  Pub/Sub 模式不適合做消息存儲(chǔ)，消息積壓類的業(yè)務(wù)慌申，而是擅長(zhǎng)處理廣播陌选，即時(shí)通訊，即時(shí)反饋的業(yè)務(wù)蹄溉。
  ?

3咨油、基于SortedSet有序集合的實(shí)現(xiàn)

ZADD KEY score member，壓入集合
ZRANGEBYSCORE柒爵，依據(jù)score獲取成員

有序集合的方案是在自己確定消息順I(yè)D時(shí)比較常用役电，使用集合成員的Score來(lái)作為消息ID，保證順序棉胀，還可以保證消息ID的單調(diào)遞增法瑟。通常可以使用時(shí)間戳+序號(hào)的方案唁奢。確保了消息ID的單調(diào)遞增霎挟，利用SortedSet的依據(jù)Score排序的特征，就可以制作一個(gè)有序的消息隊(duì)列了麻掸。
和上面的方案相比酥夭，優(yōu)點(diǎn)就是可以自定義消息ID，在消息ID有意義時(shí)脊奋，比較重要熬北。缺點(diǎn)也明顯，不允許重復(fù)消息（以為是集合）诚隙，同時(shí)消息ID確定有錯(cuò)誤會(huì)導(dǎo)致消息的順序出錯(cuò)讶隐。
?

4、基于stream實(shí)現(xiàn)

TBase還不支持該數(shù)據(jù)結(jié)構(gòu)
Redis5.0中發(fā)布的Stream類型最楷，也用來(lái)實(shí)現(xiàn)典型的消息隊(duì)列整份。該Stream類型的出現(xiàn)，幾乎滿足了消息隊(duì)列具備的全部?jī)?nèi)容火俄，包括但不限于：

• 消息ID的序列化生成
• 消息遍歷
• 消息的阻塞和非阻塞讀取
• 消息的分組消費(fèi)
• 未完成消息的處理
• 消息隊(duì)列監(jiān)控
  追加新消息讲冠，XADD，生產(chǎn)消息
  XADD谱仪，命令用于在某個(gè)stream（流數(shù)據(jù)）中追加消息否彩，演示如下：

127.0.0.1:6379> XADD memberMessage * user kang msg Hello
"1553439850328-0"
127.0.0.1:6379> XADD memberMessage * user zhong  msg nihao
"1553439858868-0"

語(yǔ)法格式為：

XADD key ID field string [field string ...]

需要提供key，消息ID方案列荔，消息內(nèi)容贴浙，其中消息內(nèi)容為key-value型數(shù)據(jù)。 ID崎溃，最常使用*，表示由Redis生成消息ID概而，這也是強(qiáng)烈建議的方案般婆。field string [field string]就是當(dāng)前消息內(nèi)容，由1個(gè)或多個(gè)key-value構(gòu)成。
上面的例子中配名，在memberMemsages這個(gè)key中追加了user kang msg Hello這個(gè)消息。Redis使用毫秒時(shí)間戳和序號(hào)生成了消息ID宇整。此時(shí)，消息隊(duì)列中就有一個(gè)消息可用了鳞青。
?
從消息隊(duì)列中獲取消息，XREAD臂拓，消費(fèi)消息
XREAD，從Stream中讀取消息胶惰，演示如下：

127.0.0.1:6379> XREAD streams memberMessage 0
1) 1) "memberMessage"
   2) 1) 1) "1553439850328-0"
         2) 1) "user"
            2) "kang"
            3) "msg"
            4) "Hello"
      2) 1) "1553439858868-0"
         2) 1) "user"
            2) "zhong"
            3) "msg"
            4) "nihao"

語(yǔ)法格式為：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

• [COUNT count] 用于限定獲取的消息數(shù)量
• [BLOCK milliseconds] 用于設(shè)置XREAD為阻塞模式，默認(rèn)為非阻塞模式
• ID 用于設(shè)置由哪個(gè)消息ID開(kāi)始讀取中捆。使用0表示從第一條消息開(kāi)始坊饶。（本例中就是使用0）此處需要注意泄伪，消息隊(duì)列ID是單調(diào)遞增的匿级，所以通過(guò)設(shè)置起點(diǎn)，可以向后讀取脓杉。在阻塞模式中简逮，可以使用無(wú)意義）悲龟。
  XRED讀消息時(shí)分為阻塞和非阻塞模式须教，使用BLOCK選項(xiàng)可以表示阻塞模式，需要設(shè)置阻塞時(shí)長(zhǎng)轻腺。非阻塞模式下贬养，讀取完畢（即使沒(méi)有任何消息）立即返回，而在阻塞模式下误算，若讀取不到內(nèi)容迷殿，則阻塞等待咖杂。
  ?
  Pending 等待列表
  為了解決組內(nèi)消息讀取但處理期間消費(fèi)者崩潰帶來(lái)的消息丟失問(wèn)題庆寺，STREAM 設(shè)計(jì)了Pending 列表，用于記錄讀取但并未處理完畢的消息翰苫。命令XPENDIING 用來(lái)獲消費(fèi)組或消費(fèi)內(nèi)消費(fèi)者的未處理完畢的消息。演示如下：

127.0.0.1:6379> XPENDING mq mqGroup # mpGroup的Pending情況
1) (integer) 5 # 5個(gè)已讀取但未處理的消息
2) "1553585533795-0" # 起始ID
3) "1553585533795-4" # 結(jié)束ID
4) 1) 1) "consumerA" # 消費(fèi)者A有3個(gè)
      2) "3"
   2) 1) "consumerB" # 消費(fèi)者B有1個(gè)
      2) "1"
   3) 1) "consumerC" # 消費(fèi)者C有1個(gè)
      2) "1"
?
127.0.0.1:6379> XPENDING mq mqGroup - + 10 # 使用 start end count 選項(xiàng)可以獲取詳細(xì)信息
1) 1) "1553585533795-0" # 消息ID
   2) "consumerA" # 消費(fèi)者
   3) (integer) 1654355 # 從讀取到現(xiàn)在經(jīng)歷了1654355ms奏窑，IDLE
   4) (integer) 5 # 消息被讀取了5次导披，delivery counter
2) 1) "1553585533795-1"
   2) "consumerA"
   3) (integer) 1654355
   4) (integer) 4
# 共5個(gè)埃唯，余下3個(gè)省略 ...
?
127.0.0.1:6379> XPENDING mq mqGroup - + 10 consumerA # 在加上消費(fèi)者參數(shù)撩匕，獲取具體某個(gè)消費(fèi)者的Pending列表
1) 1) "1553585533795-0"
   2) "consumerA"
   3) (integer) 1641083
   4) (integer) 5
# 共3個(gè)，余下2個(gè)省略 ...

每個(gè)Pending的消息有4個(gè)屬性：

• 消息ID
• 所屬消費(fèi)者
• IDLE止毕，已讀取時(shí)長(zhǎng)
• delivery counter，消息被讀取次數(shù)
  上面的結(jié)果我們可以看到漠趁，我們之前讀取的消息扁凛，都被記錄在Pending列表中，說(shuō)明全部讀到的消息都沒(méi)有處理闯传，僅僅是讀取了谨朝。那如何表示消費(fèi)者處理完畢了消息呢甥绿？使用命令XACK 完成告知消息處理完成字币，演示如下：

127.0.0.1:6379> XACK mq mqGroup 1553585533795-0 # 通知消息處理結(jié)束共缕，用消息ID標(biāo)識(shí)
(integer) 1
?
127.0.0.1:6379> XPENDING mq mqGroup # 再次查看Pending列表
1) (integer) 4 # 已讀取但未處理的消息已經(jīng)變?yōu)?個(gè)
2) "1553585533795-1"
3) "1553585533795-4"
4) 1) 1) "consumerA" # 消費(fèi)者A，還有2個(gè)消息處理
      2) "2"
   2) 1) "consumerB"
      2) "1"
   3) 1) "consumerC"
      2) "1"

有了這樣一個(gè)Pending機(jī)制图谷，就意味著在某個(gè)消費(fèi)者讀取消息但未處理后隅茎，消息是不會(huì)丟失的。等待消費(fèi)者再次上線后，可以讀取該P(yáng)ending列表降传，就可以繼續(xù)處理該消息了，保證消息的有序和不丟失出嘹。
此時(shí)還有一個(gè)問(wèn)題，就是若某個(gè)消費(fèi)者宕機(jī)之后咬崔，沒(méi)有辦法再上線了税稼，那么就需要將該消費(fèi)者Pending的消息，轉(zhuǎn)義給其他的消費(fèi)者處理垮斯，就是消息轉(zhuǎn)移郎仆。
?
消息轉(zhuǎn)移
消息轉(zhuǎn)移的操作時(shí)將某個(gè)消息轉(zhuǎn)移到自己的Pending列表中。使用語(yǔ)法XCLAIM來(lái)實(shí)現(xiàn)兜蠕，需要設(shè)置組扰肌、轉(zhuǎn)移的目標(biāo)消費(fèi)者和消息ID，同時(shí)需要提供IDLE（已被讀取時(shí)長(zhǎng)）熊杨，只有超過(guò)這個(gè)時(shí)長(zhǎng)曙旭，才能被轉(zhuǎn)移。演示如下：

# 當(dāng)前屬于消費(fèi)者A的消息1553585533795-1晶府，已經(jīng)15907,787ms未處理了
127.0.0.1:6379> XPENDING mq mqGroup - + 10
1) 1) "1553585533795-1"
   2) "consumerA"
   3) (integer) 15907787
   4) (integer) 4
?
# 轉(zhuǎn)移超過(guò)3600s的消息1553585533795-1到消費(fèi)者B的Pending列表
127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
1) 1) "1553585533795-1"
   2) 1) "msg"
      2) "2"
?
# 消息1553585533795-1已經(jīng)轉(zhuǎn)移到消費(fèi)者B的Pending中桂躏。
127.0.0.1:6379> XPENDING mq mqGroup - + 10
1) 1) "1553585533795-1"
   2) "consumerB"
   3) (integer) 84404 # 注意IDLE，被重置了
   4) (integer) 5 # 注意川陆，讀取次數(shù)也累加了1次

以上代碼剂习，完成了一次消息轉(zhuǎn)移。轉(zhuǎn)移除了要指定ID外书劝，還需要指定IDLE进倍，保證是長(zhǎng)時(shí)間未處理的才被轉(zhuǎn)移。被轉(zhuǎn)移的消息的IDLE會(huì)被重置购对，用以保證不會(huì)被重復(fù)轉(zhuǎn)移猾昆，以為可能會(huì)出現(xiàn)將過(guò)期的消息同時(shí)轉(zhuǎn)移給多個(gè)消費(fèi)者的并發(fā)操作，設(shè)置了IDLE骡苞，則可以避免后面的轉(zhuǎn)移不會(huì)成功垂蜗，因?yàn)镮DLE不滿足條件。例如下面的連續(xù)兩條轉(zhuǎn)移解幽，第二條不會(huì)成功贴见。

127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
127.0.0.1:6379> XCLAIM mq mqGroup consumerC 3600000 1553585533795-1

這就是消息轉(zhuǎn)移。至此我們使用了一個(gè)Pending消息的ID躲株，所屬消費(fèi)者和IDLE的屬性片部，還有一個(gè)屬性就是消息被讀取次數(shù)，delivery counter霜定，該屬性的作用由于統(tǒng)計(jì)消息被讀取的次數(shù)档悠，包括被轉(zhuǎn)移也算廊鸥。這個(gè)屬性主要用在判定是否為錯(cuò)誤數(shù)據(jù)上。
?
壞消息問(wèn)題辖所，Dead Letter惰说，死信問(wèn)題
正如上面所說(shuō)，如果某個(gè)消息缘回，不能被消費(fèi)者處理吆视，也就是不能被XACK，這是要長(zhǎng)時(shí)間處于Pending列表中酥宴，即使被反復(fù)的轉(zhuǎn)移給各個(gè)消費(fèi)者也是如此啦吧。此時(shí)該消息的delivery counter就會(huì)累加（上一節(jié)的例子可以看到），當(dāng)累加到某個(gè)我們預(yù)設(shè)的臨界值時(shí)幅虑，我們就認(rèn)為是壞消息（也叫死信丰滑，DeadLetter，無(wú)法投遞的消息）倒庵，由于有了判定條件褒墨，我們將壞消息處理掉即可，刪除即可擎宝。刪除一個(gè)消息郁妈，使用XDEL語(yǔ)法，演示如下：

# 刪除隊(duì)列中的消息
127.0.0.1:6379> XDEL mq 1553585533795-1
(integer) 1
# 查看隊(duì)列中再無(wú)此消息
127.0.0.1:6379> XRANGE mq - +
1) 1) "1553585533795-0"
   2) 1) "msg"
      2) "1"
2) 1) "1553585533795-2"
   2) 1) "msg"
      2) "3"

注意本例中绍申，并沒(méi)有刪除Pending中的消息因此你查看Pending噩咪，消息還會(huì)在〖模可以執(zhí)行XACK標(biāo)識(shí)其處理完畢胃碾！
?
信息監(jiān)控，XINFO
Stream提供了XINFO來(lái)實(shí)現(xiàn)對(duì)服務(wù)器信息的監(jiān)控筋搏，可以查詢仆百、查看隊(duì)列信息：

127.0.0.1:6379> Xinfo stream mq
 1) "length"
 2) (integer) 7
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "groups"
 8) (integer) 1
 9) "last-generated-id"
10) "1553585533795-9"
11) "first-entry"
12) 1) "1553585533795-3"
    2) 1) "msg"
       2) "4"
13) "last-entry"
14) 1) "1553585533795-9"
    2) 1) "msg"
       2) "10"

消費(fèi)組信息：

127.0.0.1:6379> Xinfo groups mq
1) 1) "name"
   2) "mqGroup"
   3) "consumers"
   4) (integer) 3
   5) "pending"
   6) (integer) 3
   7) "last-delivered-id"
   8) "1553585533795-4"

消費(fèi)者組成員信息：

127.0.0.1:6379> XINFO CONSUMERS mq mqGroup
1) 1) "name"
   2) "consumerA"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 18949894
2) 1) "name"
   2) "consumerB"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 3092719
3) 1) "name"
   2) "consumerC"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 23683256

?
命令一覽
?

?

Reference

[1] 基于Redis實(shí)現(xiàn)消息隊(duì)列典型方案
[2] Stream 類型