RabbitMQ消息相關

MQ的模式：

基本消息模式：一個生產者，一個消費者
work模式：一個生產者，多個消費者
訂閱模式：

????????fanout廣播模式：在Fanout模式中，一條消息，會被所有訂閱的隊列都消費。

????????在廣播模式下，消息發送流程：

可以有多個消費者
隊列的消費者都能拿到消息。實現一條消息被多個消費者消費
交換機把消息發送給綁定過的所有隊列
生產者發送的消息，只能發送到交換機，交換機來決定要發給哪個隊列，生產者無法決定。
每個隊列都要綁定到Exchange（交換機）
每個消費者有自己的queue（隊列）

????????direct定向模式：

????????在Direct下：

?隊列與交換機的綁定，不能是任意綁定了，而是要指定一個RoutingKey（路由key）
消息的發送方在向 Exchange發送消息時，也必須指定消息的 RoutingKey。
Exchange不再把消息交給每一個綁定的隊列，而是根據消息的Routing Key進行判斷，只有隊列的Routingkey與消息的 Routing key完全一致，才會接收到消息

topic通配符模式：

Topic類型的Exchange與Direct相比，都是可以根據RoutingKey把消息路由到不同的隊列。只不過Topic類型Exchange可以讓隊列在綁定Routing key 的時候使用通配符！

Routingkey 一般都是有一個或多個單詞組成，多個單詞之間以”.”分割，例如： item.insert

通配符規則：

#：匹配一個或多個詞

*：匹配不多不少恰好1個詞

舉例：

item.#：能夠匹配item.spu.insert 或者 item.spu

item.*：只能匹配item.spu

RabbitMQ 使用信道的方式來傳輸數據。信道是建立在真實的 TCP 連接內的虛擬連接，且每條 TCP 連接上的信道數量沒有限制。

單一模式；集群：普通模式（如果做了持久化，需要等待rabbit01恢復才可以被消費，如果沒有做持久化，可能造成消息丟失），鏡像模式

1、什么是RabbitMQ？為什么使用RabbitMQ？主要特性？

答：RabbitMQ是一款開源的，Erlang編寫的，基于AMQP協議的，消息中間件；

可以用它來：解耦、異步、削峰添谷。

主要特性

可伸縮性：集群服務【主從模式的】
消息持久化：從內存持久化消息到硬盤，再從硬盤加載到內存**【隊列的持久化、交換機的持久化、消息持久化】**

1.1、解耦（為面向服務的架構（SOA）提供基本的最終一致性實現）

場景說明：用戶下單后，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口。

傳統模式的缺點：

假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗
訂單系統與庫存系統耦合

引入消息隊列

訂單系統：用戶下單后，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功
庫存系統：訂閱下單的消息，采用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作
假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單后，訂單系統寫入消息隊列就不再關心其他的后續操作了。實現訂單系統與庫存系統的應用解耦
為了保證庫存肯定有，可以將隊列大小設置成庫存數量，或者采用其他方式解決。

基于消息的模型，關心的是“通知”，而非“處理”。

短信、郵件通知、緩存刷新等操作使用消息隊列進行通知。

1.2、消息隊列和RPC的區別與比較：

RPC: 異步調用，及時獲得調用結果，具有強一致性結果，關心業務調用處理結果。

消息隊列：兩次異步RPC調用，將調用內容在隊列中進行轉儲，并選擇合適的時機進行投遞（錯峰流控）

異步提升效率

場景說明：用戶注冊后，需要發注冊郵件和注冊短信。傳統的做法有兩種

1.串行的方式；2.并行方式

（1）串行方式：將注冊信息寫入數據庫成功后，發送注冊郵件，再發送注冊短信。以上三個任務全部完成后，返回給客戶端

（2）并行方式：將注冊信息寫入數據庫成功后，發送注冊郵件的同時，發送注冊短信。以上三個任務完成后，返回給客戶端。與串行的差別是，并行的方式可以提高處理的時間

（3）引入消息隊列，將不是必須的業務邏輯，異步處理。

流量削峰

流量削鋒也是消息隊列中的常用場景，一般在秒殺或團搶活動中使用廣泛。關于秒殺系列文章可以關注公眾號Java技術棧獲取閱讀。

應用場景：系統其他時間A系統每秒請求量就100個，系統可以穩定運行。系統每天晚間八點有秒殺活動，每秒并發請求量增至1萬條，但是系統最大的處理能力只能每秒處理1000個請求，于是系統崩潰，服務器宕機。

之前架構：大量用戶（100萬用戶）通過瀏覽器在晚上八點高峰期同時參與秒殺活動。大量的請求涌入我們的系統中，高峰期達到每秒鐘5000個請求，大量的請求打到MySQL上，每秒鐘預計執行3000條SQL。

但是一般的MySQL每秒鐘扛住2000個請求就不錯了，如果達到3000個請求的話可能MySQL直接就癱瘓了，從而系統無法被使用。但是高峰期過了之后，就成了低峰期，可能也就1萬用戶訪問系統，每秒的請求數量也就50個左右，整個系統幾乎沒有任何壓力。

引入MQ：100萬用戶在高峰期的時候，每秒請求有5000個請求左右，將這5000請求寫入MQ里面，系統A每秒最多只能處理2000請求，因為MySQL每秒只能處理2000個請求。

系統A從MQ中慢慢拉取請求，每秒就拉取2000個請求，不要超過自己每秒能處理的請求數量即可。MQ，每秒5000個請求進來，結果只有2000個請求出去，所以在秒殺期間（將近一小時）可能會有幾十萬或者幾百萬的請求積壓在MQ中。

這個短暫的高峰期積壓是沒問題的，因為高峰期過了之后，每秒就只有50個請求進入MQ了，但是系統還是按照每秒2000個請求的速度在處理，所以說，只要高峰期一過，系統就會快速將積壓的消息消費掉。

我們在此計算一下，每秒在MQ積壓3000條消息，1分鐘會積壓18萬，1小時積壓1000萬條消息，高峰期過后，1個多小時就可以將積壓的1000萬消息消費掉。

2、RabbitMQ有什么優缺點？

答：優點：解耦、異步、削峰；

缺點：降低了系統的穩定性：本來系統運行好好的，現在你非要加入個消息隊列進去，那消息隊列掛了，系統也就崩潰了。因此，系統可用性會降低；

增加了系統的復雜性：加入了消息隊列，要多考慮很多方面的問題，比如：一致性問題、如何保證消息不被重復消費、如何保證消息可靠性傳輸等。因此，需要考慮的東西更多，復雜性增大。

3、如何保證RabbitMQ的高可用？

答：搭建集群的RabbitMQ服務器提供服務，一臺風險太大；

4、如何保證RabbitMQ不被重復消費？

答：先說為什么會重復消費：正常情況下，消費者在消費消息的時候，消費完畢后，會發送一個確認消息給消息隊列，消息隊列就知道該消息被消費了，就會將該消息從消息隊列中刪除；

但是因為網絡傳輸等等故障，確認信息沒有傳送到消息隊列，導致消息隊列不知道自己已經消費過該消息了，再次將消息分發給其他的消費者。

【即消費者發送 ACK 前崩潰，MQ 會重新投遞消息，可能導致重復處理】

針對以上問題，一個解決思路是：保證消息的唯一性，就算是多次傳輸，不要讓消息的多次消費帶來影響；保證消息冪等性；

比如：在寫入消息隊列的數據做唯一標示（比如放到redis中），消費消息時，根據唯一標識判斷是否消費過；

ACK兩種模式

自動 ACK（Auto-Acknowledgement）

定義：消費者收到消息后，MQ 立即自動標記消息為已確認（無論消費者是否處理成功）。
特點：

- 簡單易用，但可靠性低。
- 若消費者處理消息時崩潰，消息會丟失（因為 MQ 已認為消息被確認）。

適用場景：允許少量消息丟失的非關鍵業務（如日志采集）。

手動 ACK（Manual Acknowledgement）

定義：消費者在處理消息后 顯式發送 ACK，MQ 才會刪除消息。
特點：

- 可靠性高，但需開發者主動控制。
- 若處理失敗，可發送 NACK（Negative Acknowledgement）拒絕消息，觸發重試或進入死信隊列。

適用場景：金融交易、訂單處理等對可靠性要求高的場景。

5、如何保證RabbitMQ消息的可靠傳輸(即保證消息不會丟失)？

答：消息不可靠的情況可能是消息丟失，劫持等原因；

丟失又分為：生產者丟失消息、消息隊列丟失消息、消費者丟失消息；

生產者丟失消息：【生產者發消息的可靠性】

從生產者弄丟數據這個角度來看，RabbitMQ提供transaction和confirm模式來確保生產者不丟消息；

transaction機制就是說：發送消息前，開啟事務（channel.txSelect()）,然后發送消息，如果發送過程中出現什么異常，事務就會回滾（channel.txRollback()）,如果發送成功則提交事務（channel.txCommit()）。然而，這種方式有個缺點：吞吐量下降；

confirm模式用的居多：一旦channel進入confirm模式，所有在該信道上發布的消息都將會被指派一個唯一的ID（從1開始），一旦消息被投遞到所有匹配的隊列之后；

MQ就會發送一個ACK給生產者（包含消息的唯一ID），讓生產者知道消息已經正確到達目的隊列了；

如果MQ沒能處理該消息，則會發送一個Nack消息給生產者，讓生產者進行重試操作。

處理Ack和Nack的代碼如下所示

rabbitmq:host: 192.168.112.129port: 5672username: adminpassword: pass#確認消息已發送到交換機(Exchange)#確認消息已發送到隊列(Queue)publisher-confirms: truepublisher-returns: true

消息隊列丟數據：消息持久化。【消息隊列數據的可靠性】

處理消息隊列丟數據的情況，一般是開啟持久化磁盤的配置。

這個持久化配置可以和confirm機制配合使用，你可以在消息持久化磁盤后，再給生產者發送一個Ack信號。

這樣，如果消息持久化磁盤之前，rabbitMQ陣亡了，那么生產者收不到Ack信號，生產者會自動重發。

那么如何持久化呢？

這里順便說一下吧，其實也很容易，就下面兩步declare（/d??kler/）

隊列持久化：將queue的持久化標識durable(/?d?r?bl/)設置為true，則代表是一個持久的隊列

交換機持久化

消息持久化：發送消息的時候將delivery_mode=2，// 設置消息是否持久化，1：非持久化 2：持久化

這樣設置以后，即使rabbitMQ掛了，重啟后也能恢復數據

消費者丟失消息：【消費者消費數據的可靠性】

消費者丟數據一般是因為采用了自動確認消息模式，改為手動確認消息即可！

消費者在收到消息之后，處理消息之前，會自動回復RabbitMQ已收到消息；

如果這時處理消息失敗，就會丟失該消息；

解決方案：消費者接收每一條消息后都必須進行確認（消息接收和消息確認是兩個不同操作）。只有消費者確認了消息，RabbitMQ 才能安全地把消息從隊列中刪除。

交換機、隊列、消息持久化方式總結：

為了提高并發能力，MQ的數據默認是在內存中存儲的，包括交換機、隊列、消息。

這樣就會出現數據安全問題，如果服務宕機，存儲在MQ中未被消費的消息都會丟失。

所以我們需要將交換機、隊列、消息持久化到硬盤，以防服務宕機。

交換機持久化：

隊列持久化：

消息持久化：

6、如何保證RabbitMQ消息的順序性？

答：單線程消費保證消息的順序性；對消息進行編號，消費者處理消息是根據編號處理消息；

把需要順序性消費的消息放到同一個隊列里，只允許一個消費者去消費這個隊列，不需要保證順序的就放到其他隊列里。

7、如何確保消息正確地發送至 RabbitMQ？如何確保消息接收方消費了消息？

發送方確認模式

將信道設置成 confirm 模式（發送方確認模式），則所有在信道上發布的消息都會被指派一個唯一的 ID。

一旦消息被投遞到目的隊列后，或者消息被寫入磁盤后（可持久化的消息），信道會發送一個確認給生產者（包含消息唯一 ID）。

如果 RabbitMQ 發生內部錯誤從而導致消息丟失，會發送一條 nack（notacknowledged，未確認）消息。

發送方確認模式是異步的，生產者應用程序在等待確認的同時，可以繼續發送消息。當確認消息到達生產者應用程序，生產者應用程序的回調方法就會被觸發來處理確認消息。

接收方確認機制

消費者接收每一條消息后都必須進行確認（消息接收和消息確認是兩個不同操作）。只有消費者確認了消息，RabbitMQ 才能安全地把消息從隊列中刪除。

這里并沒有用到超時機制，RabbitMQ 僅通過 Consumer 的連接中斷來確認是否需要重新發送消息。也就是說，只要連接不中斷，RabbitMQ 給了 Consumer 足夠長的時間來處理消息。保證數據的最終一致性；

下面羅列幾種特殊情況

如果消費者接收到消息，在確認之前斷開了連接或取消訂閱，RabbitMQ 會認為消息沒有被分發，然后重新分發給下一個訂閱的消費者。（可能存在消息重復消費的隱患，需要去重）

如果消費者接收到消息卻沒有確認消息，連接也未斷開，則 RabbitMQ 認為該消費者繁忙，將不會給該消費者分發更多的消息。

如何解決消息隊列的延時以及過期失效問題？消息隊列滿了以后該怎么處理？有幾百萬消息持續積壓幾小時，說說怎么解決？

消息積壓處理辦法：臨時緊急擴容：

先修復 consumer 的問題，確保其恢復消費速度，然后將現有 cnosumer 都停掉。

新建一個 topic，partition 是原來的 10 倍，臨時建立好原先 10 倍的 queue 數量。

然后寫一個臨時的分發數據的 consumer 程序，這個程序部署上去消費積壓的數據，消費之后不做耗時的處理，直接均勻輪詢寫入臨時建立好的 10 倍數量的 queue。

接著臨時征用 10 倍的機器來部署 consumer，每一批 consumer 消費一個臨時 queue 的數據。這種做法相當于是臨時將 queue 資源和 consumer 資源擴大 10 倍，以正常的 10 倍速度來消費數據。

等快速消費完積壓數據之后，得恢復原先部署的架構，重新用原先的 consumer 機器來消費消息。

MQ中消息失效：

假設你用的是 RabbitMQ，RabbtiMQ 是可以設置過期時間的，也就是 TTL。如果消息在 queue 中積壓超過一定的時間就會被 RabbitMQ 給清理掉，這個數據就沒了。那這就是第二個坑了。這就不是說數據會大量積壓在 mq 里，而是大量的數據會直接搞丟。

我們可以采取一個方案，就是批量重導，這個我們之前線上也有類似的場景干過。就是大量積壓的時候，我們當時就直接丟棄數據了，然后等過了高峰期以后，比如大家一起喝咖啡熬夜到晚上12點以后，用戶都睡覺了。這個時候我們就開始寫程序，將丟失的那批數據，寫個臨時程序，一點一點的查出來，然后重新灌入 mq 里面去，把白天丟的數據給他補回來。也只能是這樣了。假設 1 萬個訂單積壓在 mq 里面，沒有處理，其中 1000 個訂單都丟了，你只能手動寫程序把那 1000 個訂單給查出來，手動發到 mq 里去再補一次。

mq消息隊列塊滿了：如果消息積壓在 mq 里，你很長時間都沒有處理掉，此時導致 mq 都快寫滿了，咋辦？這個還有別的辦法嗎？沒有，誰讓你第一個方案執行的太慢了，你臨時寫程序，接入數據來消費，消費一個丟棄一個，都不要了，快速消費掉所有的消息。然后走第二個方案，到了晚上再補數據吧。

如何保證消息的隊列的順序型？

在生產者發送消息給隊列時，我們通過綁定多個隊列，且每個隊列都綁定唯一的消費者，消費者在消費隊列中數據時是按順序來的。

即生產者將消息存放到同一個隊列，同一個消費者在讀取該隊列中的消息是按順序讀取的。

（消息重試【補充】機制）死信隊列（Dead-Letter Queue, DLX）

一、死信隊列的定義與作用

死信隊列（DLX） 是消息隊列系統中用于存儲無法被正常消費的消息的特殊隊列。這些消息被稱為 死信（Dead-Letter Messages），通常是由于以下原因導致無法處理：

消息被消費者拒絕（Reject/Nack）且未重新入隊。
消息過期（Time-To-Live, TTL 超時）。
隊列達到最大長度限制，新消息被丟棄或舊消息被移除。
消息無法被正確路由到目標隊列（如交換機綁定錯誤）。

核心作用：

容錯處理：避免因消息處理失敗導致的無限重試或丟失。
故障排查：集中管理異常消息，便于后續分析原因。
靈活路由：支持對死信消息的再處理（如人工干預或自動重試）。

二、消息成為死信的條件

觸發條件	說明
消息被拒絕且不重入隊	消費者調用 `basic.reject` 或 `basic.nack` 并設置 `requeue=false`。
消息過期（TTL）	消息或隊列設置 TTL 超時后未被消費。
隊列滿	隊列達到最大長度（`x-max-length`）或最大容量（`x-max-length-bytes`）。
路由失敗	消息無法被交換機路由到任何隊列（需配置備用交換機 `alternate-exchange`）。

重試機制的設置

RabbitMQ自動補償機制觸發:(多用于調用第三方接口)

當我們的消費者在處理我們的消息的時候,程序拋出異常情況下觸發自動補償(默認無限次數重試)應該對我們的消息重試設置間隔重試時間,
比如消費失敗最多只能重試5次,間隔3秒(防止重復消費,冪等問題)如果重試后還未消費默認丟棄，如果配置了死信隊列，會發送到死信隊列中

listener:simple:#手動簽收acknowledge-mode: manual#并發消費者初始化值concurrency: 10#并發消費者的最大值max-concurrency: 20#每個消費者每次監聽時可拉取處理的消息數量prefetch: 5retry:#是否開啟消費者重試（為false時關閉消費者重試，這時消費端代碼異常會一直重復收到消息）enabled: true#最大重試次數max-attempts: 2#重試間隔時間（單位毫秒）initial-interval: 5000#重試最大時間間隔（單位毫秒）:當前時間間隔<max-interval（重試最大時間間隔）max-interval: 1200000#應用于前一重試間隔的乘法器:當前時間間隔=上次重試間隔*multiplier。multiplier: 2