1 MQ 存在的意義

消息中間件一般主要用來(lái)做 異步處理、應(yīng)用解耦、流量削峰、日志處理 等方面。

1.1 異步處理

一個(gè)用戶登陸網(wǎng)址注冊(cè)，然后系統(tǒng)發(fā)短信跟郵件告知注冊(cè)成功，一般有三種解決方法。

1. 串行方式，依次執(zhí)行，問(wèn)題是用戶注冊(cè)后就可以使用了，沒(méi)必要等短信跟郵件啊。

2. 注冊(cè)成功后，郵件跟驗(yàn)證碼用并行等方式執(zhí)行，問(wèn)題是郵件跟短信是非重要的任務(wù)，系統(tǒng)注冊(cè)還要等這倆完成么？

3. 基于異步MQ的處理，用戶注冊(cè)成功后直接把信息異步發(fā)送到MQ中，然后郵件系統(tǒng)跟短信系統(tǒng)主動(dòng)去消費(fèi)數(shù)據(jù)。

   
   

   

1.2 應(yīng)用解耦

比如有一個(gè)訂單系統(tǒng)，還要一個(gè)庫(kù)存系統(tǒng)，用戶下訂單后要調(diào)用庫(kù)存系統(tǒng)來(lái)處理，直接調(diào)用話，庫(kù)存系統(tǒng)出現(xiàn)問(wèn)題咋辦呢？

1.3 流量削峰

舉辦一個(gè) 秒殺活動(dòng)，如何較好到設(shè)計(jì)？服務(wù)層直接接受瞬間高密度訪問(wèn)絕對(duì)不可以，起碼要加入一個(gè)MQ來(lái)實(shí)現(xiàn)削峰。

1.4 日志處理

用戶通過(guò) WebUI 訪問(wèn)發(fā)送請(qǐng)求到時(shí)候后端如何接受跟處理呢一般？

1.5 MQ 帶來(lái)的弊端

1. 系統(tǒng)可用性降低：引入第三方依賴則需考慮第三方的穩(wěn)定性。

2. 系統(tǒng)復(fù)雜性增加：要多考慮很多方面的問(wèn)題，比如一致性問(wèn)題、如何保證消息不被重復(fù)消費(fèi)，如何保證保證消息可靠傳輸。因此需要考慮的東西更多，系統(tǒng)復(fù)雜性增大。

2 常見的 MQ

消息中間件具有低耦合、可靠投遞、廣播、流量控制、最終一致性等一系列功能，成為異步RPC的主要手段之一。當(dāng)今市面上有很多主流的消息中間件，如老牌的ActiveMQ、RabbitMQ、炙手可熱的Kafka、阿里巴巴自主開發(fā)RocketMQ等。

1. ActiveMQ：老牌的消息中間件，但是不適合高并發(fā)互聯(lián)網(wǎng)，適合傳統(tǒng)企業(yè)。

2. RabbitMQ：支持高并發(fā)、高吞吐、性能好，還有完善的管理界面等。支持集群化，缺點(diǎn)是Erlang語(yǔ)言開發(fā)的。

3. RocketMQ：阿里出品，性能優(yōu)越，Java開發(fā)，二次改造。

4. Kafka：超高吞吐量實(shí)時(shí)日志采集，一般在大數(shù)據(jù)體系配合實(shí)時(shí)計(jì)算Spark Streaming、Flink等使用。

1. 中小型公司，技術(shù)實(shí)力較為一般，技術(shù)挑戰(zhàn)不是特別高，用 RabbitMQ 是不錯(cuò)的選擇。

2. 大型公司，基礎(chǔ)架構(gòu)研發(fā)實(shí)力較強(qiáng)，用 RocketMQ 是很好的選擇。

3 RabbitMQ 常見模式

RabbitMQ 是一個(gè)開源的 AMQP 實(shí)現(xiàn)，服務(wù)器端用 Erlang 語(yǔ)言編寫，支持多種客戶端，用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性、擴(kuò)展性、高可用性等方面表現(xiàn)不俗。

AMQP：

AdvancedMessageQueuingProtocol：高級(jí)消息隊(duì)列協(xié)議，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)。

3.1 RabbitMQ 基本概念

1. Broker：簡(jiǎn)單來(lái)說(shuō)就是消息隊(duì)列服務(wù)器實(shí)體

2. Exchange：消息交換機(jī)，它指定消息按什么規(guī)則，路由到哪個(gè)隊(duì)列。

3. Queue：消息隊(duì)列載體，每個(gè)消息都會(huì)被投入到一個(gè)或多個(gè)隊(duì)列。

4. Binding：它的作用就是把 exchange 和 queue 按照路由規(guī)則綁定起來(lái)

5. Routing Key：路由關(guān)鍵字，exchange 根據(jù)這個(gè)關(guān)鍵字進(jìn)行消息投遞。

6. VHost：vhost 可以理解為虛擬 broker ，即 mini-RabbitMQ server。其內(nèi)部均含有獨(dú)立的 queue、exchange 和 binding 等，但最最重要的是，其擁有獨(dú)立的權(quán)限系統(tǒng)，可以做到 vhost 范圍的用戶控制。當(dāng)然，從 RabbitMQ 的全局角度，vhost 可以作為不同權(quán)限隔離的手段（一個(gè)典型的例子就是不同的應(yīng)用可以跑在不同的 vhost 中）。

7. Producer：消息生產(chǎn)者，就是投遞消息的程序

8. Consumer：消息消費(fèi)者，就是接受消息的程序

9. Channel：消息通道，在客戶端的每個(gè)連接里，可建立多個(gè) channel，每個(gè) channel 代表一個(gè)會(huì)話任務(wù)

由 Exchange、Queue、RoutingKey 三個(gè)才能決定一個(gè)從 Exchange 到 Queue 的唯一的線路。

3.2 RabbitMQ 工作模式

3.2.1 simple模式

簡(jiǎn)單隊(duì)列的不足：耦合性過(guò)高，生產(chǎn)者一一對(duì)應(yīng)消費(fèi)者，如果有多個(gè)消費(fèi)者想消費(fèi)隊(duì)列中信息就無(wú)法實(shí)現(xiàn)了。

3.2.2 work工作模式(資源的競(jìng)爭(zhēng))

1. 輪詢分發(fā)(round-robin)：MQ不管兩個(gè)消費(fèi)者誰(shuí)忙，數(shù)據(jù)總是你一個(gè)我一個(gè)，MQ 給兩個(gè)消費(fèi)發(fā)數(shù)據(jù)的時(shí)候是不知道消費(fèi)者性能的，默認(rèn)就是雨露均沾。此時(shí) autoAck = true。

2. 公平分發(fā)：要讓消費(fèi)者消費(fèi)完畢一條數(shù)據(jù)后就告知MQ，再讓MQ發(fā)數(shù)據(jù)即可。自動(dòng)應(yīng)答要關(guān)閉，實(shí)現(xiàn)按照消費(fèi)者性能消費(fèi)。

3.2.3 fanout publish/subscribe 發(fā)布訂閱模式

1. 一個(gè)生產(chǎn)者多個(gè)消費(fèi)者

2. 每一個(gè)消費(fèi)者都有一個(gè)自己的隊(duì)列

3. 生產(chǎn)者沒(méi)有把消息直接發(fā)送到隊(duì)列而是發(fā)送到了交換機(jī)轉(zhuǎn)化器(exchange)。

4. 每一個(gè)隊(duì)列都要綁定到交換機(jī)上。

5. 生產(chǎn)者發(fā)送的消息經(jīng)過(guò)交換機(jī)到達(dá)隊(duì)列，從而實(shí)現(xiàn)一個(gè)消息被多個(gè)消費(fèi)者消費(fèi)。

3.2.4 ?direct routing 路由模式

direct：處理路由鍵，需要指定routingKey，此時(shí)生產(chǎn)者發(fā)送數(shù)據(jù)到MQ的時(shí)候會(huì)指定key，任務(wù)隊(duì)列也會(huì)指定key，只有key一樣消息才會(huì)被傳送到隊(duì)列中。如下圖

3.2.5 topic 主題模式

將路由鍵跟某個(gè)模式匹配，生產(chǎn)者會(huì)帶 routingKey，但是消費(fèi)者的MQ會(huì)帶模糊routingKey：

1. # 表示匹配 >=1個(gè)字符。

2. * 表示匹配一個(gè)。

3. 路由功能添加模糊匹配。

4. 消息產(chǎn)生者產(chǎn)生消息，把消息交給交換機(jī)。

5. 交換機(jī)根據(jù)key的規(guī)則模糊匹配到對(duì)應(yīng)的隊(duì)列，由隊(duì)列的監(jiān)聽消費(fèi)者接收消息消費(fèi)。

3.2.6 總計(jì)

如果需要指定模式一般是在消費(fèi)者端設(shè)置，靈活性調(diào)節(jié)。

4 常見考題

4.1 消息怎么路由的

生成者生產(chǎn)消息后消息帶有 routing Key，通過(guò)routing Key 消費(fèi)者隊(duì)列被綁定到交換器上，消息到達(dá)交換器根據(jù)交換器規(guī)則匹配，常見交換器如下：

1. fanout：如果交換器收到消息，將會(huì)廣播到所有綁定的隊(duì)列上

2. direct：如果路由鍵完全匹配，消息就被投遞到相應(yīng)的隊(duì)列

3. topic：可以使來(lái)自不同源頭的消息能夠到達(dá)同一個(gè)隊(duì)列。使用 topic 交換器時(shí)，可以使用通配符

4.2 RabbitMQ 消息基于什么傳輸

信道是生產(chǎn)消費(fèi)者與rabbit通信的渠道，生產(chǎn)者 publish 或是消費(fèi)者 subscribe 一個(gè)隊(duì)列都是通過(guò)信道來(lái)通信的。信道是建立在TCP連接上的虛擬連接，就是說(shuō) RabbitMQ 在一條TCP上建立成百上千個(gè)信道來(lái)達(dá)到多個(gè)線程處理，這個(gè)TCP被多個(gè)線程共享，每個(gè)線程對(duì)應(yīng)一個(gè)信道，信道在RabbitMQ 都有唯一的ID來(lái)保證信道私有性，對(duì)應(yīng)唯一的線程使用。用信道而不用 TCP 的原因是由于 TCP 連接的創(chuàng)建和銷毀開銷較大，且并發(fā)數(shù)受系統(tǒng)資源限制，會(huì)造成性能瓶頸。

4.3 如何保證 RabbitMQ 消息不丟失

消息丟失主要分為 生產(chǎn)者丟失消息、消息列表丟失消息、消費(fèi)者丟失消息。

4.3.1 生產(chǎn)者丟失消息

RabbitMQ 提供 transaction 和 confirm 模式來(lái)確保生產(chǎn)者不丟消息。

transaction 機(jī)制就是說(shuō)：發(fā)送消息前，開啟事務(wù)(channel.txSelect)，然后發(fā)送消息，如果發(fā)送過(guò)程中出現(xiàn)什么異常，事務(wù)就會(huì)回滾（channel.txRollback()），如果發(fā)送成功則提交事務(wù)channel.txCommit()。事務(wù)卡頓會(huì)導(dǎo)致后面無(wú)法發(fā)送，官方說(shuō)加入事務(wù)機(jī)制MQ會(huì)降速250倍。

confirm(發(fā)送方確認(rèn)模式)模式用的居多：一旦 channel 進(jìn)入 confirm 模式，所有在該信道上發(fā)布的消息都將會(huì)被指派一個(gè)從1開始的唯一的ID，一旦消息被投遞到所有匹配的隊(duì)列之后，RabbitMQ 就會(huì)發(fā)送一個(gè)包含消息的唯一ID 的 ACK給生產(chǎn)者，這就使得生產(chǎn)者知道消息已經(jīng)正確到達(dá)目的隊(duì)列了，如果 RabbitMQ 沒(méi)能處理該消息，則會(huì)發(fā)送一個(gè) Nack (not acknowledged) 消息給你，你可以進(jìn)行重試操作。

發(fā)送方確認(rèn)模式是異步的，生產(chǎn)者應(yīng)用程序在等待確認(rèn)的同時(shí)，可以繼續(xù)發(fā)送消息。當(dāng)確認(rèn)消息到達(dá)生產(chǎn)者應(yīng)用程序，生產(chǎn)者應(yīng)用程序的回調(diào)方法就會(huì)被觸發(fā)來(lái)處理確認(rèn)消息。

4.3.2 消息列表 丟失消息

處理消息隊(duì)列丟數(shù)據(jù)的情況，一般是開啟持久化磁盤的配置。這個(gè)持久化配置可以和 confirm 機(jī)制配合使用，你可以在消息持久化磁盤后，再給生產(chǎn)者發(fā)送一個(gè) Ack 信號(hào)。這樣，如果消息持久化磁盤之前，RabbitMQ 掛了后生產(chǎn)者收不到Ack信號(hào)，生產(chǎn)者會(huì)自動(dòng)重發(fā)。

通過(guò)如下持久化設(shè)置，即使 RabbitMQ 掛了重啟后也能恢復(fù)數(shù)據(jù)。

1. durable = true, 將 queue 的持久化設(shè)置為 true，則代表是一個(gè)持久的隊(duì)列

2. 發(fā)送消息的時(shí)候?qū)?deliveryMode=2

關(guān)于持久化其實(shí)是個(gè)權(quán)衡問(wèn)題，持久化可能會(huì)導(dǎo)致系統(tǒng)QPS下降，所以一般僅對(duì)關(guān)鍵消息作持久化處理（根據(jù)業(yè)務(wù)重要程度），且應(yīng)該保證關(guān)鍵消息的持久化不會(huì)導(dǎo)致系統(tǒng)性能瓶頸。

4.3.3 消費(fèi)者丟失消息

消費(fèi)者丟失消息：消費(fèi)者丟數(shù)據(jù)一般是因?yàn)?strong style="font-size: inherit;line-height: inherit;color: rgb(0, 191, 165);">采用了自動(dòng)確認(rèn)消息模式，改為手動(dòng)確認(rèn)消息即可！

消費(fèi)者在收到消息之后，處理消息之前，會(huì)自動(dòng)回復(fù)RabbitMQ已收到消息；如果這時(shí)處理消息失敗，就會(huì)丟失該消息。

解決方案：處理消息成功后，手動(dòng)回復(fù)確認(rèn)消息。消費(fèi)者跟消息隊(duì)列的連接不中斷，RabbitMQ 給了 Consumer 足夠長(zhǎng)的時(shí)間來(lái)處理消息，保證數(shù)據(jù)的最終一致性。

注意點(diǎn)：

1. 消費(fèi)者接收到消息卻沒(méi)有確認(rèn)消息，連接也未斷開，則 RabbitMQ 認(rèn)為該消費(fèi)者繁忙，將不會(huì)給該消費(fèi)者分發(fā)更多的消息。

2. 如果消費(fèi)者接收到消息，在確認(rèn)之前斷開了連接或取消訂閱，RabbitMQ 會(huì)認(rèn)為消息沒(méi)有被分發(fā)，然后重新分發(fā)給下一個(gè)訂閱的消費(fèi)者，這時(shí)可能存在消息重復(fù)消費(fèi)的隱患，需要去重！

4.3 如何避免消息重復(fù)投遞或重復(fù)消費(fèi)

4.3.1 消息簡(jiǎn)介

消息重復(fù)消費(fèi)是各個(gè)MQ都會(huì)發(fā)生的常見問(wèn)題之一，在一些比較敏感的場(chǎng)景下，重復(fù)消費(fèi)會(huì)造成比較嚴(yán)重的后果，比如重復(fù)扣款等。

消息重復(fù)消費(fèi)的場(chǎng)景大概可以分為 生產(chǎn)者端重復(fù)消費(fèi) 和 消費(fèi)者端重復(fù)消費(fèi)，解決辦法是是通過(guò)冪等性來(lái)保證重復(fù)消費(fèi)的消息不對(duì)結(jié)果產(chǎn)生影響即可。

1. 消息生成時(shí) RabbitMQ 內(nèi)部 對(duì)每個(gè)生產(chǎn)的消息生成個(gè) inner-msg-id，作為去重和冪等的依據(jù)（消息投遞失敗并重傳），避免重復(fù)的消息進(jìn)入隊(duì)列。

2. 消息消費(fèi)時(shí) 要求消息體中必須要有一個(gè) bizId（對(duì)于同一業(yè)務(wù)全局唯一，如支付 ID、訂單 ID、帖子 ID 等）作為去重的依據(jù)，避免同一條消息被重復(fù)消費(fèi)。

3. 在 RocketMQ 中生產(chǎn)者發(fā)送消息前詢問(wèn) RocketMQ 信息是否已發(fā)送過(guò)，或者通過(guò)Redis記錄已查詢記錄。不過(guò)最好的還是直接在消費(fèi)端去重消費(fèi)。

4.3.2 舉例

1. 消費(fèi)者拿到這個(gè)消息做數(shù)據(jù)庫(kù)的insert操作。給這個(gè)消息做一個(gè)唯一主鍵，那么就算出現(xiàn)重復(fù)消費(fèi)的情況，就會(huì)導(dǎo)致主鍵沖突，避免數(shù)據(jù)庫(kù)出現(xiàn)臟數(shù)據(jù)。

2. 拿到消息后如果做的是 redis 的 set 操作就不用解決了，因?yàn)槟銦o(wú)論set幾次結(jié)果都是一樣的。

3. 準(zhǔn)備個(gè)第三方介質(zhì)，來(lái)做消費(fèi)記錄。以redis為例，給消息分配一個(gè)全局id，只要消費(fèi)過(guò)該消息，將以K-V形式寫入redis。那消費(fèi)者開始消費(fèi)前，先去redis中查詢有沒(méi)消費(fèi)記錄即可。

4.4 RabbitMQ 如何保證消息順序執(zhí)行

順序性 必要性：生產(chǎn)者的信息是[插入、更新、刪除]，消費(fèi)者執(zhí)行順序是[刪除、插入、更新]，這是跟預(yù)期不一致的。

4.4.1 亂序情況

出現(xiàn)消費(fèi)亂序一般是如下兩種情況：

1. 一個(gè) queue，有多個(gè) consumer 去消費(fèi)，每個(gè) consumer 的執(zhí)行時(shí)間是不固定的，無(wú)法保證先讀到消息的 consumer 一定先完成操作。
   

   

2. 一個(gè) queue 對(duì)應(yīng)一個(gè) consumer，但是 consumer 里面進(jìn)行了多線程消費(fèi)，這樣也會(huì)造成消息消費(fèi)順序錯(cuò)誤。
   

   

4.4.2 解決亂序

1. 拆分多個(gè) queue，每個(gè) queue 一個(gè) consumer，將三個(gè)有先后順序的消息根據(jù)用戶訂單id 哈希后發(fā)送到同一個(gè)queue中，來(lái)保證消息的先后性。當(dāng)然這樣會(huì)造成吞吐量下降。

2. 一個(gè) queue 對(duì)應(yīng)一個(gè) consumer，在 consumer 內(nèi)部根據(jù)ID映射到不同內(nèi)存隊(duì)列，然后用內(nèi)存隊(duì)列做排隊(duì) 分發(fā)給底層不同的 worker 來(lái)處理
   

   

4.5 RabbitMQ 的集群

RabbitMQ 是基于主從（非分布式）做高可用性的。RabbitMQ 有三種模式：單機(jī)模式、普通集群模式、鏡像集群模式。

4.5.1 單機(jī)模式

單機(jī)版的 就是 Demo 級(jí)別，生產(chǎn)系統(tǒng)一般沒(méi)人用單機(jī)模式。

4.5.2 普通集群模式

在 N 臺(tái)機(jī)器上啟動(dòng) N 個(gè) RabbitMQ 實(shí)例。創(chuàng)建的 queue 只會(huì)放在一個(gè) RabbitMQ 實(shí)例上，但每個(gè)MQ實(shí)例都 同步 queue 的元數(shù)據(jù)（元數(shù)據(jù)可以認(rèn)為是 queue 的一些配置信息，通過(guò)元數(shù)據(jù)，可以找到 queue 所在實(shí)例）。消費(fèi)時(shí)如果連接到了另外一個(gè)實(shí)例，那么那個(gè)實(shí)例會(huì)從 queue 所在實(shí)例上拉取數(shù)據(jù)過(guò)來(lái)。讓集群中多個(gè)節(jié)點(diǎn)來(lái)服務(wù)某個(gè) queue 的讀寫操作來(lái)提高吞吐量。

4.5.3 鏡像集群模式

RabbitMQ 的高可用模式，在鏡像集群模式下，你創(chuàng)建的 queue無(wú)論元數(shù)據(jù)還是 queue 里的消息都會(huì)存在于多個(gè)實(shí)例上，每個(gè) RabbitMQ 節(jié)點(diǎn)都有這個(gè) queue 的全部數(shù)據(jù)的。寫消息到 queue 的時(shí)候都會(huì)自動(dòng)把消息同步到多個(gè)實(shí)例的 queue 上。RabbitMQ 有很好的管理控制臺(tái)，就是在后臺(tái)新增一個(gè)策略，這個(gè)策略是鏡像集群模式的策略，指定的時(shí)候是可以要求數(shù)據(jù)同步到所有節(jié)點(diǎn)的，也可以要求同步到指定數(shù)量的節(jié)點(diǎn)，再次創(chuàng)建 queue 的時(shí)候，應(yīng)用這個(gè)策略，就會(huì)自動(dòng)將數(shù)據(jù)同步到其他的節(jié)點(diǎn)上去了。

1. 優(yōu)點(diǎn)在于任何一個(gè)機(jī)器宕機(jī)了其它節(jié)點(diǎn)還包含了這個(gè) queue 的完整數(shù)據(jù)，別的 consumer 都可以到其它節(jié)點(diǎn)上去消費(fèi)數(shù)據(jù)。

2. 缺點(diǎn)在于消息需要同步到所有機(jī)器上，導(dǎo)致網(wǎng)絡(luò)帶寬壓力和消耗很重。也是每個(gè)節(jié)點(diǎn)都放這個(gè) queue 的完整數(shù)據(jù)。

4.6 ?死信隊(duì)列 跟 延遲隊(duì)列

4.6.1 死信隊(duì)列

死信 Dead Letter 是 RabbitMQ 中的一種消息機(jī)制，當(dāng)消費(fèi)消息時(shí)隊(duì)列里的消息出現(xiàn)以下情況那么該消息將成為死信。死信消息會(huì)被RabbitMQ進(jìn)行特殊處理，如果配置了 死信隊(duì)列 信息，那么該消息將會(huì)被丟進(jìn)死信隊(duì)列中，如果沒(méi)有配置，則該消息將會(huì)被丟棄：

1. 消息被否定確認(rèn)，使用channel.basicNack 或 channel.basicReject ?，并且此時(shí) default-requeue-rejected(由于監(jiān)聽器拋出異常而拒絕的消息是否被重新放回隊(duì)列) 屬性被設(shè)置為false。

2. 消息在隊(duì)列的存活時(shí)間超過(guò)設(shè)置的TTL時(shí)間。

3. 消息隊(duì)列的消息數(shù)量已經(jīng)超過(guò)最大隊(duì)列長(zhǎng)度。

1. 對(duì)隊(duì)列中消息總數(shù)進(jìn)行限制，x-max-length = 指定值。則超出閾值后隊(duì)頭數(shù)據(jù)被拋棄。

2. 對(duì)隊(duì)列中消息體總字節(jié)數(shù)進(jìn)行限制，只計(jì)算消息體的字節(jié)數(shù)。x-max-length-bytes = 指定值。

死信隊(duì)列 并不是什么特殊的隊(duì)列，只不過(guò)是綁定在死信交換機(jī)上的隊(duì)列。死信交換機(jī)只不過(guò)是用來(lái)接受死信的普通交換機(jī)，所以可以為任何類型，比如Direct、Fanout、Topic。

適用場(chǎng)景：

在較為重要的業(yè)務(wù)隊(duì)列中，確保未被正確消費(fèi)的消息不被丟棄，在系統(tǒng)因?yàn)閰?shù)解析、數(shù)據(jù)校驗(yàn)、網(wǎng)咯撥打等導(dǎo)致異常后通過(guò)配置死信隊(duì)列，可以讓未正確處理的消息暫存到另一個(gè)隊(duì)列中，待后續(xù)排查清楚問(wèn)題后，編寫相應(yīng)的處理代碼來(lái)處理死信消息。

死信消息的生命周期：

1. 業(yè)務(wù)消息被投入業(yè)務(wù)隊(duì)列

2. 消費(fèi)者消費(fèi)業(yè)務(wù)隊(duì)列的消息，由于處理過(guò)程中發(fā)生異常，于是進(jìn)行了nck或者reject操作

3. 被nck或reject的消息由RabbitMQ投遞到死信交換機(jī)中

4. 死信交換機(jī)將消息投入相應(yīng)的死信隊(duì)列

5. 死信隊(duì)列的消費(fèi)者消費(fèi)死信消息

死信消息是 RabbitMQ 為我們做的一層保證，其實(shí)我們也可以不使用死信隊(duì)列，而是在消息消費(fèi)異常時(shí)，將消息主動(dòng)投遞到另一個(gè)交換機(jī)中，明白死信隊(duì)列運(yùn)行機(jī)制后就知道這些 Exchange 和 Queue 想怎樣配合就能怎么配合。比如從死信隊(duì)列拉取消息，然后發(fā)送郵件、短信、釘釘通知來(lái)通知開發(fā)人員關(guān)注?；蛘邔⑾⒅匦峦哆f到一個(gè)隊(duì)列然后設(shè)置過(guò)期時(shí)間，來(lái)進(jìn)行延時(shí)消費(fèi)。

4.6.2 RabbitMQ 中的 TTL

TTL(Time To Live) 是 RabbitMQ 中一個(gè) 消息 或 隊(duì)列 的屬性，如果一條消息設(shè)置了 TTL屬性或者進(jìn)入了有 TTL屬性的隊(duì)列，那么這條消息如果在TTL設(shè)置的時(shí)間內(nèi)沒(méi)有被消費(fèi)，則會(huì)成為死信。如果同時(shí)配置了隊(duì)列的TTL和消息的TTL，那么較小的那個(gè)值將會(huì)被使用。

1. queue 設(shè)置 TTL

1Map?args?=?new?HashMap();
2args.put("x-message-ttl",?6000);?//?ms
3channel.queueDeclare(queueName,?durable,?exclusive,?autoDelete,?args);

2. Msg 設(shè)置 TTL

1AMQP.BasicProperties.Builder?builder?=?new?AMQP.BasicProperties.Builder();
2builder.expiration("6000");
3AMQP.BasicProperties?properties?=?builder.build();
4channel.basicPublish(exchangeName,?routingKey,?mandatory,?properties,?"msg?body".getBytes());

區(qū)別：

1. 設(shè)置了隊(duì)列的TTL屬性，一旦Msg 過(guò)期，就會(huì)被隊(duì)列丟棄。

2. Msg 設(shè)置 TTL，Msg 是否過(guò)期是在即將投遞到消費(fèi)者之前判定的，如果當(dāng)前隊(duì)列有嚴(yán)重的Msg 積壓情況，則已過(guò)期的 Msg 也許還能存活較長(zhǎng)時(shí)間，解決辦法 安裝插件 rabbitmq_delayed_message_exchange。

3. 如果不設(shè)置TTL，表示 Msg 永遠(yuǎn)不會(huì)過(guò)期，TTL = 0 表示除非此時(shí)可以直接投遞該 Msg 到消費(fèi)者，否則該 Msg 將會(huì)被丟棄。

4.6.3 延遲隊(duì)列

延時(shí)隊(duì)列中的元素則是希望被在指定時(shí)間得到取出和處理，所以延時(shí)隊(duì)列中的元素是都是帶時(shí)間屬性的，通常來(lái)說(shuō)是需要被處理的消息或者任務(wù)。一般用在如下場(chǎng)景：

1. 訂單在 15 分鐘之內(nèi)未支付則自動(dòng)取消。

2. 賬單在一周內(nèi)未支付，則自動(dòng)結(jié)算。

3. 用戶注冊(cè)成功后，如果三天內(nèi)沒(méi)有登陸則進(jìn)行短信提醒。

4. 用戶發(fā)起退款，如果三天內(nèi)沒(méi)有得到處理則通知相關(guān)運(yùn)營(yíng)人員。

5. 預(yù)定會(huì)議后，需要在預(yù)定的時(shí)間點(diǎn)前十分鐘通知各個(gè)與會(huì)人員參加會(huì)議。

   延時(shí)隊(duì)列 = 死信隊(duì)列 ?+ ?TTL
   

   

   
   

   
   

6. 當(dāng)然也可以用 Java 的 DelayQueue、Quartz、Redis 的 zset 等實(shí)現(xiàn)。

4.7 MQ 消息積壓咋辦

這種時(shí)候只能操作臨時(shí)擴(kuò)容，以更快的速度去消費(fèi)數(shù)據(jù)了。具體操作步驟和思路如下：

1. 先修復(fù)consumer的問(wèn)題，確保其恢復(fù)消費(fèi)速度，然后將現(xiàn)有consumer都停掉。

2. 臨時(shí)建立好原先10倍~20倍的queue數(shù)量(新建一個(gè)topic，partition是原來(lái)的10倍)。

3. 然后寫一個(gè)臨時(shí)分發(fā)消息的 consumer 程序，這個(gè)程序部署上去消費(fèi)積壓的消息，消費(fèi)之后不做耗時(shí)處理，直接均勻輪詢寫入臨時(shí)建好分10數(shù)量的queue里面。

4. 緊接著征用10倍的機(jī)器來(lái)部署 consumer，每一批 consumer消費(fèi)一個(gè)臨時(shí) queue 的消息。

5. 這種做法相當(dāng)于臨時(shí)將 queue 資源和 consumer 資源擴(kuò)大10倍，以正常速度的10倍來(lái)消費(fèi)消息。

6. 等快速消費(fèi)完了之后，修復(fù)consumer，去消費(fèi)新的MQ和現(xiàn)有的MQ數(shù)據(jù)，新MQ消費(fèi)完成后恢復(fù)原狀。
   

   

4.8 RabbitMQ 中的推拉

在RabbitMQ 中有推模式跟拉模式，平時(shí)開發(fā)多為推模式。

1. 推模式：消息中間件主動(dòng)將消息推送給消費(fèi)者

2. 拉模式：消費(fèi)者主動(dòng)從消息中間件拉取消息

4.8.1 推模式 push

1. 推模式接收消息是最有效的一種消息處理方式。channel.basicConsume(queneName,consumer)方法將信道(channel)設(shè)置成投遞模式，直到取消隊(duì)列的訂閱為止。當(dāng)消息到達(dá)RabbitMQ時(shí)，RabbitMQ會(huì)自動(dòng)地、不斷地投遞消息給匹配的消費(fèi)者，而不需要消費(fèi)端手動(dòng)來(lái)拉取，當(dāng)然投遞消息的個(gè)數(shù)還是會(huì)受到channel.basicQos的限制。

2. 推模式將消息提前推送給消費(fèi)者，消費(fèi)者必須設(shè)置一個(gè)緩沖區(qū)緩存這些消息。優(yōu)點(diǎn)是消費(fèi)者總是有一堆在內(nèi)存中待處理的消息，所以當(dāng)真正去消費(fèi)消息時(shí)效率很高。缺點(diǎn)就是緩沖區(qū)可能會(huì)溢出。

3. 由于推模式是信息到達(dá)RabbitMQ后，就會(huì)立即被投遞給匹配的消費(fèi)者，所以實(shí)時(shí)性非常好，消費(fèi)者能及時(shí)得到最新的消息。

4.8.2 拉模式 pull

1. 如果只想從隊(duì)列中獲取單條消息而不是持續(xù)訂閱，則可以使用channel.basicGet方法來(lái)進(jìn)行消費(fèi)消息。

2. 拉模式在消費(fèi)者需要時(shí)才去消息中間件拉取消息，這段網(wǎng)絡(luò)開銷會(huì)明顯增加消息延遲，降低系統(tǒng)吞吐量。

3. 由于拉模式需要消費(fèi)者手動(dòng)去RabbitMQ中拉取消息，所以實(shí)時(shí)性較差；消費(fèi)者難以獲取實(shí)時(shí)消息，具體什么時(shí)候能拿到新消息完全取決于消費(fèi)者什么時(shí)候去拉取消息。

4.9 設(shè)計(jì)個(gè)MQ

一般是個(gè)開放題，考察有沒(méi)有從架構(gòu)角度整體構(gòu)思和設(shè)計(jì)的思維以及能力。不求看過(guò)源碼起碼但的知道基本原理、核心組成部分、基本架構(gòu)構(gòu)成，然后參照一些開源的技術(shù)把一個(gè)系統(tǒng)設(shè)計(jì)出來(lái)的思路說(shuō)一下就好（強(qiáng)行為下篇Kafka做鋪墊）。

1. 考慮MQ的伸縮性，在需要的時(shí)候快速擴(kuò)容來(lái)增加吞吐量和容量，設(shè)計(jì)個(gè)分布式的系統(tǒng)，參照一下kafka的設(shè)計(jì)理念，broker、 topic、 partition，每個(gè)partition放一個(gè)機(jī)器，就存一部分?jǐn)?shù)據(jù)。如果現(xiàn)在資源不夠了，給topic增加partition，然后做數(shù)據(jù)遷移，增加機(jī)器，提供更高的吞吐量了。

2. 落磁盤方式為順序?qū)?/strong>，這樣就沒(méi)有磁盤隨機(jī)讀寫的尋址開銷，磁盤順序讀寫的性能是很高的，這就是Kafka的思路。

3. 參考Kafka實(shí)現(xiàn)MQ高可用性，多副本 -> leader & follower -> broker 掛了重新選舉leader即可對(duì)外服務(wù)。

4. 參考前面的實(shí)現(xiàn)數(shù)據(jù)的零丟失。

參考

1. 消息順序消費(fèi)：https://www.jianshu.com/p/02fdcb9e8784

2. RabbitMQ死信隊(duì)列：https://www.cnblogs.com/mfrank/category/1514703.html

3. RabbitMQ消息推送：https://blog.csdn.net/qq_40837310/article/details/109033000