工作中經常用到消息中間件來解決系統間的解耦問題或者高并發消峰問題，但是消息的可靠性如何保證一直是個很大的問題，什么情況下消息就不見了？如何防止消息丟失？下面通過這篇文章，我們就聊聊RabbitMQ 消息可靠性如何解決的？

本文分三部分說明

RabbitMQ 消息丟失場景有哪些？

如何避免消息丟失？

如何設計部署消息中間件保證消息可靠性？

RabbitMQ 消息丟失場景有哪些？

首先我們看下消息周期投遞過程：

我們把該圖分三部分，左中右，每部分都會導致消息丟失情況,下面就詳細聊聊每個階段消息是如何丟的：

1.生產者生產消息到RabbitMQ Server 消息丟失場景

1) 外界環境問題導致：發生網絡丟包、網絡故障等造成RabbitMQ Server端收不到消息，因為生產環境的網絡是很復雜的，網絡抖動，丟包現象很常見，下面會講到針對這個問題是如何解決的。

2) 代碼層面，配置層面，考慮不全導致消息丟失

事例1:
一般情況下，生產者使用Confirm模式投遞消息，如果方案不夠嚴謹，比如RabbitMQ Server 接收消息失敗后會發送nack消息通知生產者，生產者監聽消息失敗或者沒做任何事情，消息存在丟失風險；

事例2:
生產者發送消息到exchange后，發送的路由和queue沒有綁定，消息會存在丟失情況，下面會講到具體的例子，保證意外情況的發生，即使發生，也在可控范圍內。

2.RabbitMQ Server中存儲的消息丟失或可靠性不足

1)消息未完全持久化，當機器重啟后，消息會全部丟失，甚至Queue也不見了

假如：你僅僅持久化了Message,而Exchange,Queue沒有持久化，這個持久化是無效的。記得之前公司有一哥們忘記持久化Queue導致機器重啟后，Queue不見了，自然Message也丟失了。

2)單節點模式問題，如果某個節點掛了，消息就不能用了,業務可能癱瘓，只能等待

如果做了消息持久化方案，消息會持久化硬盤，機器重啟后消息不會丟失；但是還有一個極端情況，這臺服務器磁盤突然壞了(公司遇到過磁盤問題還是很多的)，消息持久化不了，非高可用狀態，這個模式生產環境慎重考慮。

3)普通集群模式：某個節點掛了，該節點上的消息不能用，有影響的業務癱瘓，只能等待節點恢復重啟可用(建立在消息持久化)

雖然這個模式進步了一點點，多個節點，但是消息還是不能保證可靠，為什么呢？

因為RabbitMQ 集群模式有點特殊，隊列的內容僅僅存在某一個節點上面，不會存在所有節點上面，所有節點僅僅存放消息結構和元數據(可以理解為索引，這也是為了提高性能，如果每次把所有內容同步到所有節點是有開銷代價的)。下面自己畫了一張圖介紹普通集群丟失消息情況：

這里有三個節點，通常情況下一個磁盤節點，兩個內存節點，首先先說明下， Queue1 內容僅僅存在節點note1上面，在創建隊列的時候已經固定了，note2,note3 僅僅存放的是元數據，這個一定要清楚，Producer發送消息到note2，note2 會同步元數據到其他節點，內容會同步note1。

那我們想下，圖中的Q1問題，note1掛了，這個節點的Queues全部暫時不可用，節點恢復后可用。

我們說下圖片中備注2中的問題，Producer發送消息到note2，note2在同步note1前note1掛了，此時你的心情是怎么樣的。。。后面會講具體的策略

4)鏡像模式：可以解決上面的問題，但是還是有意外情況發生

比如：持久化的消息，保存到硬盤過程中，當前隊列節點掛了，存儲節點硬盤又壞了，消息丟了，怎么辦？下面會詳細介紹

3.RabbitMQ Server到消費者消息丟失

消費端接收到相關消息之后，消費端還沒來得及處理消息，消費端機器就宕機了，此時消息如果處理不當會有丟失風險，后面會講到如何處理這個情況，消費端也有ack機制

如何避免消息丟失？

下面也是從三個方面介紹：

1.生產者生產消息到RabbitMQ Server 可靠性保證

2.RabbitMQ Server中存儲的消息如何保證

3.RabbitMQ Server到消費者消息如何不丟

1. 生產者生產消息到RabbitMQ Server可靠性保證

這個過程，消息可能會丟，比如發生網絡丟包、網絡故障等造成消息丟失，一般情況下如果不采取措施，生產者無法感知消息是否已經正確無誤的發送到exchange中,如果生產者能感知到的話，它可以進行進一步的處理動作，比如重新投遞相關消息以確保消息的可靠性。

1.1 通常有一種方案可以解決：就是 AMQP協議提供的一個事務機制

RabbitMQ客戶端中Channel 接口提供了幾個事務機制相關的方法：

channel.txSelect

channel.txCommit

channel.txRollback

源碼截圖如下：com.rabbitmq.client 包中public interface Channel extendsShutdownNotifier {}接口

在生產者發送消息之前，通過channel.txSelect開啟一個事務，接著發送消息， 如果消息投遞server失敗，進行事務回滾channel.txRollback，然后重新發送， 如果server收到消息，就提交事務channel.txCommit

但是，很少有人這么干，因為這是同步操作，一條消息發送之后會使發送端阻塞，以等待RabbitMQ Server的回應，之后才能繼續發送下一條消息，生產者生產消息的吞吐量和性能都會大大降低。

1.2 幸運的是RabbitMQ提供了一個改進方案，即發送方確認機制(publisher confirm)

首先生產者通過調用channel.confirmSelect方法將信道設置為confirm模式，一旦信道進入confirm模式，所有在該信道上面發布的消息都會被指派一個唯一的ID(從1開始)，一旦消息被投遞到所有匹配的隊列之后，RabbitMQ就會發送一個確認(Basic.Ack)給生產者(包含消息的唯一deliveryTag和multiple參數)，這就使得生產者知曉消息已經正確到達了目的地了。

其實Confirm模式有三種方式實現：

串行confirm模式：producer每發送一條消息后，調用waitForConfirms()方法，等待broker端confirm，如果服務器端返回false或者在超時時間內未返回，客戶端進行消息重傳。

批量confirm模式：producer每發送一批消息后，調用waitForConfirms()方法，等待broker端confirm。

異步confirm模式：提供一個回調方法，broker confirm了一條或者多條消息后producer端會回調這個方法。我們分別來看看這三種confirm模式

串行confirm

for(int i = 0;i<50;i++){

channel.basicPublish(

exchange, routingKey,

mandatory, immediate,

messageProperties,

message.getContent()

);

if (channel.waitForConfirms()) {

System.out.println("發送成功");

} else {

//發送失敗這里可進行消息重新投遞的邏輯

System.out.println("發送失敗");

}

}

批量confirm模式

for(int i = 0;i<50;i++){

channel.basicPublish(

exchange, routingKey,

mandatory, immediate,

messageProperties,

message.getContent()

);

}

if (channel.waitForConfirms()) {

System.out.println("發送成功");

} else {

System.out.println("發送失敗");

}

上面代碼是簡單版本的，生產環境絕對不是循環發送的，而是根據業務情況， 各個客戶端程序需要定期(每x秒)或定量(每x條)或者兩者結合來publish消息，然后等待服務器端confirm。相比普通confirm模式，批量可以極大提升confirm效率。

但是有沒有發現什么問題？

問題1: 批量發送的邏輯復雜化了。

問題2: 一旦出現confirm返回false或者超時的情況時，客戶端需要將這一批次的消息全部重發，這會帶來明顯的重復消息數量，并且當消息經常丟失時，批量confirm性能應該是不升反降的。

異步confirm模式

Channel channel = channelManager.getPublisherChannel(namespaceName);

ProxiedConfirmListener confirmListener = new ProxiedConfirmListener();//監聽類

confirmListener.setChannelManager(channelManager);

confirmListener.setChannel(channel);

confirmListener.setNamespace(namespaceName);

confirmListener.addSuccessCallbacks(successCallbacks);

channel.addConfirmListener(confirmListener);

channel.confirmSelect();//開啟confirm模式

AMQP.BasicProperties messageProperties = null;

if (message.getProperty() instanceof AMQP.BasicProperties) {

messageProperties = (AMQP.BasicProperties) message.getProperty();

}

confirmListener.toConfirm(channel.getNextPublishSeqNo(), rawMsg);

for(int i = 0;i<50;i++){

channel.basicPublish(

exchange, routingKey,

mandatory, immediate,

messageProperties,

message.getContent()

);

}

異步模式需要自己多寫一部分復雜的代碼實現，異步監聽類，監聽server端的通知消息，異步的好處性能會大幅度提升，發送完畢之后，可以繼續發送其他消息。MQServer通知生產端ConfirmListener監聽類：用戶可以繼承接口實現自己的實現類，處理消息確認機制，此處繼承類代碼省略，就是上面 ProxiedConfirmListener 類：下面貼下要實現的接口：

package com.rabbitmq.client;

import java.io.IOException;

/**

* Implement this interface in order to be notified of Confirm events.

* Acks represent messages handled successfully; Nacks represent

* messages lost by the broker. Note, the lost messages could still

* have been delivered to consumers, but the broker cannot guarantee

* this.

*/

public interface ConfirmListener {

/**

** handleAck RabbitMQ消息接收成功的方法，成功后業務可以做的事情

** 發送端投遞消息前，需要把消息先存起來，比如用KV存儲，接收到ack后刪除

**/

void handleAck(long deliveryTag, boolean multiple)

throws IOException;

//handleNack RabbitMQ消息接收失敗的通知方法，用戶可以在這里重新投遞消息

void handleNack(long deliveryTag, boolean multiple)

throws IOException;

}

上面的接口很有意思，如果是你的話，怎么實現？消息投遞前如何存儲消息，ack 和 nack 如何處理消息？

下面看下異步confirm的消息投遞流程：

解釋下這張圖片：

channel1 連續發類1，2，3條消息到RabbitMQ-Server，RabbitMQ-Server通知返回一條通知，里面包含回傳給生產者的確認消息中的deliveryTag包含了確認消息的序號，此外還有一個參數multiple=true，表示到這個序號之前的所有消息都已經得到了處理。這樣客戶端和服務端通知的次數就減少類，提升類性能。

channel3 發送的消息失敗了，生產端需要對投遞消息重新投遞，需要額外處理代碼。那么生產端需要做什么事情呢？因為是異步的，生產端需要存儲消息然后根據server通知的消息，確認如何處理，于是我們面臨的問題是：

第一：發送消息之前把消息存起來

第二：監聽ack 和 nack 并做響應處理

那么怎么存儲呢？

我們分析下，可以使用SortedMap 存儲，保證有序，但是有個問題高并發情況下， 每秒可能幾千甚至上萬的消息投遞出去，消息的ack要等幾百毫秒的話，放內存可能有內存溢出的風險。所以建議采用KV存儲，KV存儲承載高并發能力高，性能好，但是要保證KV 高可用，單個有個缺點就是又引入了第三方中間件，復雜度升高。

解決了上面的問題，下面還會遇到一個問題，消息丟失的另一個情況？

事務機制和publisher confirm機制確保的是消息能夠正確的發送至RabbitMQ，這里的“發送至RabbitMQ”的含義是指消息被正確的發往至RabbitMQ的交換器，如果此交換器沒有匹配的隊列的話，那么消息也將會丟失，怎么辦？

這里有兩個解決方案，

1. 使用mandatory 設置true

2. 利用備份交換機(alternate-exchange)：實現沒有路由到隊列的消息

我們看下RabbitMQ客戶端代碼方法

Channel 類中 發布消息方法

void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)

throws IOException;

解釋下：basicPublish 方法中的，mandatory和immediate

/**

* 當mandatory標志位設置為true時，如果exchange根據自身類型和消息routeKey無法找到一個符合條件的queue， 那么會調用basic.return方法將消息返回給生產者

* 當mandatory設置為false時，出現上述情形broker會直接將消息扔掉。

*/

@Setter(AccessLevel.PACKAGE)

private boolean mandatory = false;

/**

* 當immediate標志位設置為true時，如果exchange在將消息路由到queue(s)時發現對于的queue上沒有消費者， 那么這條消息不會放入隊列中。

當immediate標志位設置為false時,exchange路由的隊列沒有消費者時，該消息會通過basic.return方法返還給生產者。

* RabbitMQ 3.0版本開始去掉了對于immediate參數的支持，對此RabbitMQ官方解釋是：這個關鍵字違背了生產者和消費者之間解耦的特性，因為生產者不關心消息是否被消費者消費掉

*/

@Setter(AccessLevel.PACKAGE)

private boolean immediate;

所以為了保證消息的可靠性，需要設置發送消息代碼邏輯。如果不單獨形式設置mandatory=false

使用mandatory 設置true的時候有個關鍵點要調整，生產者如何獲取到沒有被正確路由到合適隊列的消息呢？通過調用channel.addReturnListener來添加ReturnListener監聽器實現，只要發送的消息，沒有路由到具體的隊列，ReturnListener就會收到監聽消息。

channel.addReturnListener(new ReturnListener() {

public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP

.BasicProperties basicProperties, byte[] body) throws IOException {

String message = new String(body);

//進入該方法表示，沒路由到具體的隊列

//監聽到消息，可以重新投遞或者其它方案來提高消息的可靠性。

System.out.println("Basic.Return返回的結果是：" + message);

}

});

此時有人問了，不想復雜化生產者的編程邏輯，又不想消息丟失，那么怎么辦？還好RabbitMQ提供了一個叫做alternate-exchange東西，翻譯下就是備份交換器，這個干什么用呢？很簡單，它可以將未被路由的消息存儲在另一個exchange隊列中，再在需要的時候去處理這些消息。

那如何實現呢？

簡單一點可以通過webui管理后臺設置，當你新建一個exchange業務的時候，可以給它設置Arguments，這個參數就是 alternate-exchange，其實alternate-exchange就是一個普通的exchange，類型最好是fanout 方便管理

當你發送消息到你自己的exchange時候，對應key沒有路由到queue，就會自動轉移到alternate-exchange對應的queue，起碼消息不會丟失。

下面一張圖看下投遞過程：

那么有人有個疑問，上面介紹了，兩種方式處理，發送的消息無法路由到隊列的方案， 如果備份交換器和mandatory參數一起使用，會有什么效果？

答案是：mandatory參數無效

由于篇幅太長，我會再分一篇文章出來講下面的內容

2. RabbitMQ Server中存儲的消息如何保證消息可靠性和高可用

待續...

3. RabbitMQ Server到消費者消息如何不丟

待續...

再聊聊大廠都是如何使用在生產環境的

待續...

END

如有收獲，請幫忙轉發，后續會有更好文章貢獻，您的鼓勵是作者最大的動力！

歡迎長按下圖關注公眾號?技術工匠精神

總結

以上是生活随笔為你收集整理的rabbitmq可靠性投递_解决RabbitMQ消息丢失问题和保证消息可靠性（一）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。