??用redis也是比較久了，并且也對(duì)其他消息中間件也用了相當(dāng)多的時(shí)間，現(xiàn)在就redis是否適合做消息隊(duì)列來梳理下，獲取梳理完之后，可以有一個(gè)更加清晰的認(rèn)知。筆者會(huì)從以下幾個(gè)方面進(jìn)行梳理。

一、簡(jiǎn)單的list消息隊(duì)列

??眾所周知，redis常見的數(shù)據(jù)結(jié)構(gòu)有String、Hash、List、Set、zset。其中List可以是一個(gè)列表結(jié)構(gòu)?？梢酝ㄟ^LPUSH、RPOP兩個(gè)命令來實(shí)現(xiàn)一個(gè)簡(jiǎn)單的隊(duì)列。

• LPUSH 將元素依次插入到列表頭部
• RPOP 獲取最后一個(gè)元素，并且刪除。

??如下圖所示：生產(chǎn)者通過LPUSH命令，依次插入a、b、c、d四個(gè)元素。消費(fèi)者通過RPOP命令進(jìn)行消費(fèi)。

1.命令示例

生產(chǎn)者：

# 通過LPUSH命令往test_queue填充a、b、c、d
127.0.0.1:6379> LPUSH test_queue a
(integer) 1
127.0.0.1:6379> LPUSH test_queue b
(integer) 2
127.0.0.1:6379> LPUSH test_queue c
(integer) 3
127.0.0.1:6379> LPUSH test_queue d
(integer) 4
127.0.0.1:6379>

? 消費(fèi)者：

消費(fèi)者
127.0.0.1:6379> RPOP test_queue
"a"
127.0.0.1:6379> RPOP test_queue
"b"
127.0.0.1:6379> RPOP test_queue
"c"
127.0.0.1:6379> RPOP test_queue
"d"
127.0.0.1:6379> RPOP test_queue
(nil)
127.0.0.1:6379>

2.偽代碼示例

??生產(chǎn)者相對(duì)簡(jiǎn)單，以簡(jiǎn)單的訂單支付為例子

//在訂單支付成功之后發(fā)送給積分系統(tǒng)
public void afterPayHandler(Order order){//發(fā)送消息到積分系統(tǒng)redisTemp.LPUSH("order",order);
}

消費(fèi)者

public void orderMessageListener(){while(true){//獲取訂單信息Order order = redisTemp.RPOP("order");if(order != null){//做積分系統(tǒng)的業(yè)務(wù)，如添加積分等邏輯。}}
}

??如上所示：消費(fèi)者在消費(fèi)的時(shí)候，必須通過循環(huán)一直拉取隊(duì)列數(shù)據(jù)，達(dá)到數(shù)據(jù)的實(shí)時(shí)性，但是也出現(xiàn)了CPU空轉(zhuǎn)的問題。如果我們判斷空的時(shí)候sleep休眠一段時(shí)間，那就會(huì)存在消息實(shí)時(shí)性問題。休眠多久合適就成為了難以處理的問題。

好在redis有阻塞拉取的命令。

BRPOP test_queue 10(秒)。拉取命令，阻塞10秒。如果是0就是一直阻塞。

3.方案優(yōu)劣

1. 優(yōu)點(diǎn)：足夠簡(jiǎn)單，也很好理解。但是我確實(shí)是想不到哪個(gè)場(chǎng)景適合這個(gè)方案（笑哭）。感覺也只能算普及知識(shí)了。
2. 缺點(diǎn)：
   1. 不支持多消費(fèi)者。任何一個(gè)消費(fèi)者將redis的元素拉取刪除之后，其他消費(fèi)者都無法再次拉取到。那就只能僅限于一對(duì)一消費(fèi)了。
   2. 消息丟失。沒有ACK機(jī)制，如果拉取消息后宕機(jī)后，無法正常消費(fèi)，就會(huì)導(dǎo)致消息的丟失。

二、Pub/Sub發(fā)布訂閱

??List數(shù)據(jù)結(jié)構(gòu)可以認(rèn)為是開發(fā)者為了簡(jiǎn)單方便，從而引進(jìn)的一種消息隊(duì)列的方式，但是絕不”正宗“。Pub/Sub這種從名字上可以看出來，就是專門為了消息隊(duì)列而生的。

??? 從上圖可以看出，發(fā)布訂閱模式，解決了多消費(fèi)者的問題。但是還是存在兩個(gè)問題。

1.消息丟失

??發(fā)布訂閱模型，沒有進(jìn)行消息存儲(chǔ)，只是一個(gè)單純的通道，實(shí)時(shí)的把消息傳送給消費(fèi)者。那么這樣就會(huì)有一個(gè)問題，如果消費(fèi)者中間下線，再次上線的時(shí)候，只能從最新的位置進(jìn)行消費(fèi)，這樣就會(huì)有消息丟失啦。

2.消息堆積

??上文說，發(fā)布訂閱模型沒有基于任何數(shù)據(jù)類似，因此，這個(gè)操作不會(huì)寫入RDB和AOF中（redis持久化機(jī)制）。另外，在消息堆積的時(shí)候，數(shù)據(jù)是通過Buffer緩沖區(qū)實(shí)現(xiàn)的。這個(gè)緩沖區(qū)的大小可以在redis中進(jìn)行配置。如果超過了緩沖區(qū)配置的上限，此時(shí)，Redis 就會(huì)「強(qiáng)制」把這個(gè)消費(fèi)者踢下線。

??總的說，這個(gè)發(fā)布訂閱模式相對(duì)比較脆弱，雖然解決了多消費(fèi)者的問題，但是消息一致性較低，消息丟失概率較高（發(fā)布版本時(shí)重啟了就可能丟消息），試用的場(chǎng)景較少。

三、相對(duì)成熟的Stream

??Redis5.0 中增加了Stream消息隊(duì)列相對(duì)成熟，解決了較多的問題。

1. 支持消息ACK反饋，在消息消費(fèi)成功的時(shí)候，返回消費(fèi)成功，才不會(huì)再次推送消息。
2. 支持多消費(fèi)者組。
3. 消息堆積問題優(yōu)化。

1.redis命令介紹

發(fā)布命令

解釋 
#topic為 myStream1 
# * 代表使用自動(dòng)生成的ID作為消息的ID
# 接下來是多個(gè) field value 組成的信息。
127.0.0.1:6379> XADD myStream1 * name zhangsan sex 20
"1717500637523-0"
127.0.0.1:6379> XADD myStream1 * name lisi sex 20
"1717500644429-0"
127.0.0.1:6379>

消費(fèi)命令

# 消費(fèi)myStream隊(duì)列的10個(gè)數(shù)據(jù)，最后的0意思是從頭開始消費(fèi)。
127.0.0.1:6379> XREAD COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717500637523-0"2) 1) "name"2) "zhangsan"3) "sex"4) "20"2) 1) "1717500644429-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>

2.多消費(fèi)者組測(cè)試

#創(chuàng)建一個(gè)消費(fèi)者組為myGroup1并且指定消費(fèi)位置。最后這個(gè)長(zhǎng)串是信息的id
127.0.0.1:6379> XGROUP CREATE myStream1 myGroup1 1717500644429-0
OK

#消費(fèi)者組消費(fèi)，消費(fèi)者組為myGroup1 當(dāng)前消費(fèi)者id為consumer1 拉取10個(gè)信息  注意最后這個(gè) ‘>‘
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 >
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>

驗(yàn)證ACK機(jī)制

1. myGroup1消費(fèi)者組拉取一次之后將所有的消息拉取回來
2. 因?yàn)闆]有進(jìn)行消息反饋ACK。所以再次拉取的時(shí)候，還是將全量的消息拉取回來。
3. 執(zhí)行一次ACK命令之后，再次拉取消息，發(fā)現(xiàn)少了一條消息。
4. 再次執(zhí)行ACK命令后，拉取不到消息了。

127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"2) 1) "1717502213457-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"2) 1) "1717502213457-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379>
127.0.0.1:6379> XACK myStream1 myGroup1 1717502213457-0
(integer) 1
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) 1) 1) "1717501299234-0"2) 1) "name"2) "lisi"3) "sex"4) "20"
127.0.0.1:6379> XACK myStream1 myGroup1 1717501299234-0
(integer) 1
127.0.0.1:6379> XREADGROUP GROUP myGroup1 consumer1 COUNT 10 STREAMS myStream1 0
1) 1) "myStream1"2) (empty list or set)
127.0.0.1:6379>

3.Stream會(huì)持久化嗎？

??會(huì)，不管是RDB 還是AOF都會(huì)寫入。所以不用擔(dān)心宕機(jī)的問題。

4.消息堆積如何解決？

??既然會(huì)將Stream會(huì)進(jìn)行持久化，那么必然消息也會(huì)保存在內(nèi)存中，但是為了內(nèi)存爆炸，Stream可以在XADD命令的時(shí)候，可以通過MAXLEN命令指定消息的最大長(zhǎng)度，在超過最大長(zhǎng)度的時(shí)候，舊消息會(huì)被刪除，只保留固定長(zhǎng)度的新消息。這樣看來，消息堆積的問題只是進(jìn)行了優(yōu)化，并沒有完美的解決。

總結(jié)

??到此，獲取對(duì)于redis消息隊(duì)列的歷史有了一定的了解，redis作為運(yùn)行在內(nèi)存的數(shù)據(jù)庫而言，這個(gè)功能已經(jīng)是很不錯(cuò)了，或許你的場(chǎng)景足夠簡(jiǎn)單，消息的數(shù)量不多，并且對(duì)于消息的丟失不是特別的敏感的話，redis的Stream消息隊(duì)列也是一個(gè)不錯(cuò)的選擇。