柚子快報邀請碼778899分享：RabbitMQ

http://yzkb.51969.com/

在消息隊列（MQ）中，確保消息成功傳遞是一個關鍵問題。消息傳遞的過程通常包括以下幾個階段：publisher（生產者） -> exchange（交換機） -> queue（隊列） -> consumer（消費者）。為了確保消息在每個階段都能成功傳遞，我們需要采取一系列措施來保證消息的可靠性。

生產者的可靠性

重試機制

當生產者與交換機（或隊列，如果沒有交換機）之間的連接不穩(wěn)定時，生產者發(fā)送的消息可能會在傳輸過程中丟失。在這種情況下，生產者需要等待一段時間以獲取響應。如果未收到響應，生產者應嘗試重新發(fā)送消息。重試次數(shù)應有限制，以防止因持續(xù)重試而占用過多資源。此外，重試之間應有一定的間隔時間，以避免頻繁重試導致資源浪費。

由于發(fā)送消息時會占用通道，其他業(yè)務操作可能會被阻塞，直到消息發(fā)送完成（無論成功或失?。?。因此，對發(fā)送消息的重試機制進行限制是必要的，以防止因連接問題導致資源被長時間占用。

以下是一個在Spring Boot中配置生產者重試機制的示例：

spring:

rabbitmq:

connection-timeout: 1s # 連接超時時間

template:

retry:

enabled: true # 開啟重試機制

initial-interval: 1000ms # 初始重試間隔時間

multiplier: 1 # 重試間隔時間的倍數(shù)

max-attempts: 5 # 最大重試次數(shù)

在這個配置中：

connection-timeout?設置了連接超時的時間。 enabled: true?開啟了重試機制。 initial-interval: 1000ms?設置了在連接失敗后，首次重試前的等待時間。 multiplier: 1?設置了每次重試后等待時間的倍數(shù)（在此例中，等待時間保持不變）。 max-attempts: 5?設置了最大重試次數(shù)，超過該次數(shù)后將不再重試。

通過這種配置，生產者在發(fā)送消息失敗后會自動進行重試，直到達到最大重試次數(shù)或消息成功發(fā)送。這種機制有效地提高了消息傳遞的可靠性，同時避免了因持續(xù)重試而導致的資源浪費

根據(jù)您提供的信息，我們可以分析MQ連接失敗時的重試行為。以下是詳細的分析：

連接超時時間：設置為1秒。這意味著如果MQ在1秒內未能成功連接，連接嘗試將被視為失敗。 初始重試間隔：設置為1秒。在第一次連接失敗后，系統(tǒng)會等待1秒再進行下一次連接嘗試。 等待時間倍數(shù)：設置為1。這意味著每次重試的等待時間保持不變。因此，每次重試的間隔時間為1秒（等待時間） + 1秒（連接超時時間） = 2秒。 最大重試次數(shù)：設置為5次。系統(tǒng)會在達到最大重試次數(shù)后停止嘗試連接。

根據(jù)這些設置，系統(tǒng)在連接失敗后的行為如下：

第一次連接失敗后，等待1秒，然后進行第二次連接嘗試。 第二次連接失敗后，再次等待1秒，然后進行第三次連接嘗試。 第三次連接失敗后，系統(tǒng)將停止嘗試連接。

如果multiplier設置為2，重試行為將有所不同：

第一次連接失敗后，等待1秒，然后進行第二次連接嘗試。 第二次連接失敗后，等待時間翻倍為2秒，然后進行第三次連接嘗試。 第三次連接失敗后，等待時間再次翻倍為4秒，然后系統(tǒng)將停止嘗試連接。

這種配置確保了系統(tǒng)在連接失敗時能夠進行有限次數(shù)的重試，同時通過調整等待時間倍數(shù)來控制重試的頻率，以避免過度占用資源。

確認機制

在正常情況下，消息傳遞到MQ后不會發(fā)生丟失，但我們仍需對消息丟失有所防備。為了及時發(fā)現(xiàn)消息丟失，MQ通常使用Publisher Confirm和Publisher Return兩種機制來進行預警。

消息丟失的可能情況

MQ內部故障：MQ服務器內部出現(xiàn)問題，導致消息丟失。 交換機或隊列不存在：消息無法找到目標交換機或隊列。 RoutingKey不匹配：消息的路由鍵（RoutingKey）沒有匹配的隊列。 其他異常情況：例如網(wǎng)絡故障、消息過期（TTL）等。

消息傳遞的幾種狀態(tài)

路由失敗，但投遞成功：

消息傳遞到MQ后，MQ服務器成功接收并存儲了消息（投遞成功）。 但由于路由失?。ɡ鏡outingKey不匹配或目標隊列不存在），消息無法被正確路由到下一個節(jié)點（例如隊列）。 MQ服務器會返回ACK確認投遞成功，同時通過Publisher Return機制返回路由失敗的異常信息。 臨時消息投遞成功：

臨時消息傳遞到MQ后，被存儲在內存中。 MQ服務器返回ACK，表示消息已成功投遞。 持久化消息投遞成功：

持久化消息傳遞到MQ后，被持久化存儲到磁盤。 MQ服務器返回ACK，表示消息已成功投遞。 投遞失?。?/p>

如果消息未能成功傳遞到MQ（例如網(wǎng)絡故障或MQ服務器不可用），MQ服務器會返回NACK，表示投遞失敗。

Publisher Confirm 和 Publisher Return 的作用

Publisher Confirm：

通過返回ACK或NACK，告知生產者消息是否成功投遞到MQ服務器。 ACK表示消息已成功投遞（無論是臨時消息還是持久化消息）。 NACK表示消息投遞失敗。 Publisher Return：

當消息成功投遞到MQ服務器但路由失敗時，通過Publisher Return返回異常信息。 幫助生產者及時發(fā)現(xiàn)消息無法被正確路由的問題。

總結

投遞成功：消息成功到達MQ服務器并被存儲（臨時消息存儲在內存中，持久化消息存儲在磁盤中），MQ返回ACK。 路由失敗：消息成功投遞到MQ服務器，但無法被正確路由到目標隊列，MQ通過Publisher Return返回異常信息。 投遞失?。合⑽茨艹晒鬟f到MQ服務器，MQ返回NACK。

通過Publisher Confirm和Publisher Return機制，生產者可以及時了解消息的投遞狀態(tài)，從而有效預防和發(fā)現(xiàn)消息丟失問題。

spring: ? rabbitmq: ? ? publisher-confirm-type: correlated? ? ? publisher-returns: true?

默認情況下, Publisher Confirm和Publisher Return兩個機制是關閉的. publisher-confirm-type有三鐘狀態(tài)

none: 默認關閉, 就是這個狀態(tài)

simple: 同步進行, 需要等到回復狀態(tài)之后才會繼續(xù)業(yè)務

correlated: 異步進行, 在等待回復狀態(tài)的同時, 業(yè)務可以繼續(xù)進行處理

MQ的可靠性

數(shù)據(jù)持久化

確保消息隊列（MQ）可靠性的關鍵措施

在分布式系統(tǒng)中，消息隊列（MQ）是異步通信的核心組件。然而，即使在正常情況下，消息在到達MQ后仍有可能丟失。因此，確保MQ的可靠性至關重要。本文將介紹幾種常見的MQ可靠性方案。

1. 消息丟失的原因

MQ通常將消息存儲在內存中進行處理和傳遞，這種方式雖然高效，但在MQ服務重啟或崩潰時，內存中的消息會丟失。為了解決這一問題，我們需要采取以下措施來增強MQ的可靠性。

2. MQ可靠性方案

2.1 交換機的持久化

交換機（Exchange）是MQ中路由消息的關鍵組件。通過將交換機設置為持久化，可以確保在MQ重啟后，交換機的配置和元數(shù)據(jù)不會丟失，從而保證消息能夠繼續(xù)被正確路由。

2.2 隊列的持久化

隊列（Queue）是消息的存儲載體。將隊列設置為持久化后，即使MQ服務重啟，隊列中的消息也不會丟失。持久化隊列會將消息存儲到磁盤中，而不是僅僅依賴內存。

2.3 消息的持久化

除了交換機和隊列的持久化，消息本身也需要進行持久化處理。持久化的消息在未被消費前會一直存儲在磁盤中，只有在被成功消費后才會被刪除。這種方式可以有效避免因MQ重啟或崩潰導致的消息丟失。

3. 生產者確認機制

為了進一步增強MQ的可靠性，可以啟用生產者確認機制（Publisher Confirms）。當消息被持久化存儲到磁盤后，MQ會向生產者發(fā)送一個確認（ACK），告知消息已安全存儲。這種機制可以確保消息不會在傳輸過程中丟失。

4. 批量持久化與異步處理

為了提高性能，MQ通常不會逐條持久化消息，而是采用批量持久化的方式。這種方式可以顯著減少磁盤I/O操作，提升系統(tǒng)的整體效率。同時，推薦使用異步方式進行持久化，以避免阻塞消息的處理流程。

5. 總結

通過交換機的持久化、隊列的持久化、消息的持久化以及生產者確認機制，可以顯著提升MQ的可靠性。此外，批量持久化和異步處理能夠在不犧牲可靠性的前提下，進一步提高系統(tǒng)的性能。在實際應用中，建議根據(jù)業(yè)務需求合理配置這些機制，以確保消息的可靠傳遞。

LazyQueue

消息存儲在內存中的優(yōu)勢與挑戰(zhàn)

RabbitMQ 默認將消息存儲在內存中，因為內存的讀寫速度遠高于硬盤，這可以顯著提高消息處理的效率。然而，這種設計也帶來了一些潛在的問題，尤其是在消息量激增的情況下。

1. 內存存儲的優(yōu)勢

高效讀寫：?內存的訪問速度比硬盤快得多，因此將消息存儲在內存中可以大幅降低消息收發(fā)的延遲，提升系統(tǒng)性能。 低延遲：?對于實時性要求較高的場景，內存存儲能夠確保消息快速傳遞。

2. 內存存儲的挑戰(zhàn)

盡管內存存儲有諸多優(yōu)勢，但在某些情況下，可能會面臨以下問題：

2.1 生產者消息激增

當生產者的消息發(fā)送速率突然增加時，可能會導致消息在內存中大量堆積。

2.2 消費者處理能力不足

如果消費者的處理速度跟不上生產者的發(fā)送速度，消息會在內存中積壓，占用大量內存資源。

2.3 內存限制與 PageOut

內存的容量是有限的，當消息積壓超過內存的極限時，RabbitMQ 會將部分消息從內存轉移到硬盤中，這個過程稱為?PageOut。在 PageOut 過程中：

生產者可能會被拒絕發(fā)送消息（流控機制生效）。 消費者也無法消費消息，因為部分消息正在從內存轉移到硬盤。 內存資源會被占用，影響系統(tǒng)的整體性能。

3. Lazy Queue 的引入

為了解決上述問題，RabbitMQ 在?3.12 版本?之后引入了?Lazy Queue?機制。Lazy Queue 的核心設計思想是：

消息優(yōu)先存儲到硬盤：?消息不會直接存儲在內存中，而是批量寫入硬盤。 懶加載機制：?只有當消費者需要消費消息時，才會將消息從硬盤加載到內存中。 支持海量消息存儲：?Lazy Queue 可以輕松處理百萬級甚至更多的消息量，而不會對內存造成過大壓力。

4. Lazy Queue 的優(yōu)勢

降低內存壓力：?消息主要存儲在硬盤中，內存占用大幅減少。 提高系統(tǒng)穩(wěn)定性：?即使消息量激增，也不會因為內存不足而導致消息丟失或系統(tǒng)崩潰。 適合高吞吐量場景：?對于消息量大但實時性要求不高的場景，Lazy Queue 是一個理想的選擇。

消費者的可靠性

當RabbitMq將消息傳遞給消費者, 依舊會存在之前的消息丟失, 比如消費者處理異常, 發(fā)送失敗, 消費者宕機等情況

消費者確認機制

當RabbitMQ將消息傳遞給消費者時，仍然可能存在消息丟失的情況，例如消費者處理異常、發(fā)送失敗、消費者宕機等。為了確保消息的可靠傳遞，RabbitMQ提供了消費者確認機制，允許消費者在處理消息后向RabbitMQ反饋消息的處理狀態(tài)。

消費者確認機制

RabbitMQ的消費者確認機制主要分為以下三種狀態(tài)：

ack（確認）：消息處理成功，RabbitMQ會將該消息從隊列中刪除。 nack（否定確認）：消息處理異常，RabbitMQ會將消息重新加載回隊列，進行重試。 reject（拒絕）：消息處理異常，且消息無法被處理，RabbitMQ會直接刪除該消息。reject通常用于消息類型不匹配或無法處理的場景。

Spring AMQP中的確認模式

Spring AMQP為開發(fā)者提供了三種確認模式，以簡化消息確認的處理：

none模式：

當消息發(fā)送到消費者后，RabbitMQ會立即確認（ack）并刪除消息，無論消費者是否成功處理。 適用于對消息丟失不敏感的場景。 manual模式：

開發(fā)者需要手動調用ack、nack或reject來確認消息的處理狀態(tài)。 適用于需要精細控制消息確認的場景，但可能會造成業(yè)務代碼的污染。 auto模式：

Spring AMQP會自動根據(jù)消息處理的結果發(fā)送確認。

如果業(yè)務正常處理，返回ack。 如果發(fā)生業(yè)務異常，返回nack，消息會重新入隊進行重試。 如果發(fā)生消息校驗或處理異常，返回reject，消息會被直接刪除。 適用于大多數(shù)常見的業(yè)務場景。

配置示例

在Spring Boot中，可以通過以下配置來設置確認模式：

spring:

rabbitmq:

listener:

simple:

acknowledge-mode: auto # 可選值為 none, manual, auto

異常處理的最佳實踐

在auto模式下，建議根據(jù)異常類型決定是返回nack還是reject。例如，對于可重試的業(yè)務異常（如網(wǎng)絡抖動），可以返回nack；對于不可恢復的異常（如消息格式錯誤），可以返回reject。

失敗重試機制

業(yè)務異常處理與重試機制

在消息隊列中，如果不對業(yè)務異常處理進行合理的限制，每次業(yè)務發(fā)生異常時，消息會重新入隊并進行重試。如果沒有適當?shù)目刂茩C制，消息可能會不斷重試，導致系統(tǒng)資源被空耗，甚至引發(fā)消息堆積、系統(tǒng)崩潰等問題。

在Spring AMQP中，可以通過配置RabbitMQ的監(jiān)聽器重試機制來避免這種情況。以下是一個典型的配置示例：

spring:

rabbitmq:

listener:

simple:

retry:

enabled: true # 開啟重試機制

max-attempts: 5 # 最大重試次數(shù)

initial-interval: 1000 # 初始重試間隔時間（毫秒）

multiplier: 1 # 重試間隔時間的倍數(shù)

stateless: true # 是否啟用無狀態(tài)重試

配置項說明：

enabled: 是否啟用重試機制。 max-attempts: 最大重試次數(shù)。 initial-interval: 初始重試間隔時間。 multiplier: 重試間隔時間的倍數(shù)。 stateless: 是否啟用無狀態(tài)重試。設置為true時，表示每次重試都是獨立的，不會保留前一次重試的狀態(tài)。這對于一些涉及上下文代碼或變量傳遞的業(yè)務場景非常重要。如果業(yè)務邏輯依賴于某些變量的初始狀態(tài)，啟用無狀態(tài)重試可以確保每次重試時變量都恢復到初始值，避免前一次重試的結果影響后續(xù)業(yè)務。

無狀態(tài)重試的重要性

在某些業(yè)務場景中，業(yè)務邏輯可能會依賴于某些變量的初始狀態(tài)。例如：

int x = 5; // 初始值為5

// 業(yè)務邏輯處理

x = 8; // 業(yè)務處理后，x的值變?yōu)?

如果在下一次重試時，變量x仍然保留上一次處理后的值（即8），而不是恢復到初始值（5），這可能會導致業(yè)務邏輯出現(xiàn)錯誤。通過設置stateless: true，可以確保每次重試時，業(yè)務邏輯中的變量都會恢復到初始狀態(tài)，從而避免這種問題。

失敗處理策略

當消息在本地多次重試失敗后, 超過重試次數(shù)的限制, 會被隊列刪除, 但是這對于消息可靠性要求高的業(yè)務并不友好, 所以提供了一個MessageRecovery接口, 這個接口較好的實現(xiàn)是RepublishMessageRecovery類, 它是把異常信息放到一個單獨的隊列中, 后續(xù)人工介入處理

業(yè)務冪等性

在執(zhí)行某個業(yè)務操作時，無論該操作被執(zhí)行一次還是多次，最終的業(yè)務結果都是一致的，這就是冪等性。冪等性在分布式系統(tǒng)和網(wǎng)絡通信中尤為重要，因為它可以有效避免重復操作帶來的問題。

舉個例子：當你在購物時進行扣款操作，如果由于網(wǎng)絡延遲或其他原因，扣款請求被重復發(fā)送了兩次，而你被扣了兩次錢，這顯然是不可接受的。為了避免這種情況，我們需要確?？劭畈僮鞯膬绲刃?。

唯一消息ID

在消息隊列系統(tǒng)中，為了確保消息的冪等性（即多次處理同一消息不會產生重復的效果），通常會在發(fā)送消息時附帶一個唯一ID。這個唯一ID可以是全局唯一標識符（UUID）或根據(jù)業(yè)務規(guī)則生成的唯一值。消費者在接收到消息后，首先會檢查該ID是否已經存在于數(shù)據(jù)庫中。如果該ID已經存在，說明該消息已經被處理過，消費者會直接跳過該消息，避免重復處理；如果該ID不存在，消費者則會處理該消息，并將該ID存儲到數(shù)據(jù)庫中，以確保后續(xù)重復消息不會被重復處理。

業(yè)務判斷

除了使用唯一消息ID來確保消息的冪等性外，業(yè)務判斷也是處理重復消息或請求的重要手段。業(yè)務判斷是通過業(yè)務本身的邏輯來進行判斷，確保即使消息ID不同，但業(yè)務內容相同的請求也不會被重復處理。例如，在訂單系統(tǒng)中，可以通過訂單號、用戶ID等業(yè)務字段來判斷是否已經處理過相同的請求。

通過結合唯一消息ID和業(yè)務判斷，可以有效地確保消息隊列系統(tǒng)中的消息處理是冪等的，從而避免重復處理帶來的業(yè)務問題。

兜底方案

盡管MQ消息系統(tǒng)已經盡可能減少了消息丟失的可能性，但在實際應用中，消息丟失的情況仍然可能發(fā)生。因此，我們需要主動采取措施來確保業(yè)務結果的準確性。具體來說，可以通過定時任務來實現(xiàn)這一目標。定時任務會每隔一段時間對業(yè)務結果進行比對和檢查，確保數(shù)據(jù)的完整性和一致性。

這種兜底方案的核心在于通過定期檢查來彌補消息丟失可能帶來的影響，從而保證業(yè)務的最終一致性。

消息冪等與性能優(yōu)化

在高數(shù)據(jù)量的場景下，消息冪等性不僅要保證消息處理的正確性，還需要提升消息的存取速度和系統(tǒng)效率。以下是幾種常見的優(yōu)化方案：

集群部署：通過集群化部署，可以提高系統(tǒng)的穩(wěn)定性和吞吐量，分散單節(jié)點的壓力，從而提升整體性能。 分庫分表：對于數(shù)據(jù)量巨大的場景，可以采用分庫分表的方式，將數(shù)據(jù)分散到多個數(shù)據(jù)庫或表中，減少單表的數(shù)據(jù)量，提升查詢和寫入效率。 數(shù)據(jù)生命周期管理：通過定時任務對數(shù)據(jù)進行歸檔、移動或刪除，確保數(shù)據(jù)的時效性，避免無效數(shù)據(jù)占用存儲資源，從而提升系統(tǒng)性能。

這些方法結合使用，可以在保證消息冪等性的同時，有效提升系統(tǒng)的處理能力和效率

延時消息

在網(wǎng)上售賣貨物時，由于商品數(shù)量有限，當用戶下單后，數(shù)據(jù)庫會自動扣減商品庫存。然而，如果用戶未在規(guī)定時間內完成付款，這些商品就會被該用戶占用，導致其他購物者無法購買。為了解決這一問題，我們需要對未付款的用戶設置時間限制，要求他們在規(guī)定時間內完成付款，否則庫存將被釋放。

這種需要在一定時間后再執(zhí)行的任務被稱為延時任務。消息隊列（MQ）提供了兩種處理延時任務的解決方案：死信交換機和延時消息插件。

死信交換機

死信是指以下幾種情況下的消息：

超時未被處理的消息：消息在隊列中等待時間過長，未被消費者處理。 隊列滿員：當隊列達到最大容量時，新消息無法進入，成為死信。 消費失敗的消息：消費者在處理消息時返回nack或reject，并且將requeue設置為false，表示消息消費失敗且不再重新入隊。

當一個隊列中存在死信消息時，可以通過配置dead-letter-exchange參數(shù)，將這些死信消息轉發(fā)到一個特定的交換機，這個交換機被稱為死信交換機。死信交換機會與一個或多個隊列綁定，用于接收和處理這些死信消息。

延時消息插件

延時消息插件是一種簡化消息延時處理的工具，相較于使用死信交換機的方案，它減少了實現(xiàn)延時消息所需的步驟。延時消息插件在基本的消息流程（生產者 -> 交換機 -> 隊列 -> 消費者）中，通過將交換機設計為具備延時和暫時存儲消息的能力，從而直接實現(xiàn)消息的延時投遞。這種方式避免了傳統(tǒng)死信交換機方案中需要額外設置死信隊列、綁定死信交換機等復雜操作，簡化了系統(tǒng)的設計和維護。

柚子快報邀請碼778899分享：RabbitMQ

http://yzkb.51969.com/