rabbitMq中消息可靠性的示例分析

這篇文章主要為大家展示了“rabbitMq中消息可靠性的示例分析”，內容簡而易懂，條理清晰，希望能夠幫助大家解決疑惑，下面讓小編帶領大家一起研究并學習一下“rabbitMq中消息可靠性的示例分析”這篇文章吧。

正文： 消息雖然有異步、高并發的優點，但具有一定延時、數據弱一致性，甚至消息發送/消費失敗等缺點。所以，我們可以從生產者端，服務器和消費者端來保證消息可靠性，如果這三端都能保證，消息可靠性也就OK了。

生產者端：

1. 異常捕獲機制 客戶端代碼中的異常捕獲，如果在業務邏輯執行過程中拋出異常，則重發或者回滾。 但是，沒有異常不代表一定成功，所以此方案是最大努力保證。

注：也可以通過spring.rabbitmq.template.retry.enabled=true 配置開啟發送端的重試

2. AMQP/RabbitMQ的事務機制 開啟mq的事務模式，如果事務能成功提交，肯定發送成功了。 如下圖：

但是，此方式性能開銷很大，一般不用。

3. 發送端確認機制（publish confirm，后文簡稱pc) 所謂發送端消息確認，就是保證消息發送成功，如果不成功要能知道哪些消息不成功，并做出相應的處理。

pc原理： 在創建channel后，把channel設置為confirm模式，通過此信道發送的消息都會生成一個唯一id（msg_id，從1開始）。消息成功發送到隊列后，會給發送端返回ack確認消息，失敗則返回Nack消息。  

特別說明： 所謂返回ack和nack，其實就是兩個回調函數。 回調函數中包含msg_id，還有兩個參數：  Long deliveryTag 和Boolean multiple。  如果multiple為true，代表msg_id <= deliveryTag 的消息都被確認了（Nack代表都沒確認）； 如果multiple為false，代表只有msg_id=deliverTag的這條消息被確認了（Nack代表沒被確認）。

下面，分三種情況來說一下pc模式的應用（前兩種做了解不推薦，引出第三種）。

confirm模式1：

以上這種方式是同步阻塞，發送一條消息等到結果確認后，才發送下一條，性能很差。

confirm模式2： 在模式1的基礎上，采用批量發送。

此模式和模式1一樣也是同步，通過異常判斷，只不過是一批等待一次，效率好一些。 不過最大的問題是，如果失敗了，我們只知道這一批有問題，但不知道是哪一些，無法針對性處理。

confirm模式3： 還記得我們前面說ack和nack回調函數，但是前兩種模式都沒有顯示的調用，只是通過異常來判斷。。我們可以通過重寫和顯示調用，來處理成功或失敗消息的邏輯，這樣，生產者發送消息后不需要等待結果，可繼續投遞下一條。

如上圖（也是需要先把channel聲明為確認模式，圖中沒有截出來），定義重寫了兩個ConfirmCallback，在addConfirmListener中根據參數來區分，誰是ack，誰是nack。 當消息成功或失敗后會執行相關的回調，可以在其中處理邏輯，如每次發送消息后，會把消息也放到ConcurrentNavigableMap中，當執行ack回調后，，清空map中已被確認的消息（ConcurrentNavigableMap的用法，自己去補哈）。 服務器端：  消息持久化。持久化是保障mq可靠的基礎，有幾個方面： Exchange： 定義時設置durable 參數為true Queue：定義時設置durable 參數為true 消息： 投遞模式設置為2（BasicProperties 中的 deliveryMode 屬性）

除此之外，如有需要，broker端還需要做高可用的集群架構。  這里不做描述，后面會專門拿一篇來講如何做。 消費者端： 消費者端要保證的就是，消息一定消費成功，如果不成功，消息不能丟失。 可以從三方面來做： 1、采用NONE模式，即不需要ack確認 。  但是一定要在代碼中把消息存起來，并作異常捕獲，確認成功后再刪除。 2、采用AUTO（自動Ack）模式。    如果有異常會把消息重新放回隊列，重新消費直到成功或者過期。  但是，沒異常也不代表消費一定會成功。 3、采用MANUAL（手動Ack）模式。   手動確認是保證消息不丟失的最好途徑。

二： 可靠性其他優化 單從消息層面來說，保證發送端，消費端和服務器的可靠后，基本就問題不大了。。但是從系統層面來說，還是需要再做一些優化： 如果消息生成遠超過消費，比如秒殺瞬間消息量暴增，消息中間件本身的緩沖能力是有容量限制的，就可能導致broker崩潰。 限流，可以從幾個方面來做： 限流： 1、rabbit可以對內存和磁盤使用量設置閾值，達到閾值后，會暫時阻止消息發布者端的連接，不再接收消息。 可在/etc/rabbitmq/rabbitmq.conf中配置： 磁盤配置: 內存配置：

2、基于credit flow 的流控機制，這個機制是針對單個connection的。 在單個隊列達到最大流速，或該connection下所有隊列達到總流速時，都會觸發流控。 觸發時可能時因為connetion，exchange，或者是queue的某一個過程處于flow狀態，這些可以通過界面查看到： 3、限制channel上未被ack確認的消息數 通過在消費前，調用channel.basicQoS 方法設置該數量-prefetchCount。  一旦給消費的發送的消息達到prefetchCount未確認，就不會再向消費者推送了 （非None Ack模式下的推送模式 才有效果），這樣可以緩解消費者壓力。

4、另外，可以提升消費端的消費能力，如優化程序性能，增加消費節點，增加并發消費線程數等。

二：消息的可靠性分析

 在使用中間件過程中，我們有上面很多的方式來保證可靠性，可也難保有消息丟失的情況，這時我們需要有消息軌跡可以追蹤。

rabbitmq可以使用Firehose 功能來追蹤生產者發送或者消費者消費的消息，原理是把需要追蹤的消息，通過一個topic類型的交換機amq.rabbitmq.trace發送到隊列中（首先要去建立兩個隊列，一個放發送的消息，一個放消費的消息，然后將此交換機和兩個隊列綁定，bingdingKey分別為：publish.{exchangename} 和 deliver. {queuename}。exchangename和queuename代表正常業務中，生產者發送消息的交換器名稱 和 消費者消費消息的隊列名稱）。 綁定后，在正常發送或消費消息，這些消息就能路由到剛才兩個追蹤的隊列中。

使用Firehose，需要先開啟Firehose命令 （默認關閉，服務重啟恢復默認狀態），開啟和關閉命令如下：

rabbitmqctl trace_on [-p vhost]        rabbitmqctl trace_off [-p vhost]

Firehose 開啟之后多少會影響RabbitMQ 整體服務性能，因為它會引起額 外的消息生成、路由和存儲。此外，沒有管理界面，只能獲取追蹤隊列的消息來看，不方便。

這時候，我們可以使用rabbitmq_tracing 插件，它是Firehose的GUI版本，可以通過界面方便的操作，能把消息以text/json的格式記錄到文件中，方便追蹤。 用法如下：

1、開啟插件： rabbitmq-plugins enable rabbitmq_tracing   （禁用是disable）

2、在界面中操作（不需要創建隊列，是保存在文件中）  比如要記錄發送的消息：

如圖，在Add a new trace可以創建一個新的trace（追蹤）:  name: 生成trace文件的名稱，自定義 format: 消息保存文件的形式  text 或 json username/password： 連接rabbit的用戶名和密碼 payload bytes： 消息的大小，超過這個數后不能記錄到文件中。 不填即所有消息 pattern： 指定哪些 發送方/消費方的消息會記錄到文件（可用通配符）。 如：        publish.#：代表只要是投遞的消息，都要記錄。        deliver.#:  代表只要是消費的信息，都要記錄。        #.amq： 代表交換機名稱為amq的投遞消息，或者隊列名為amq的消費消息,都要記錄。

新建trace后，可用在界面停止采集或刪除，也可以查看里面的消息內容。    這種用法，對于一些特別重要的消息，保證可靠性還是很有必要的

以上是“rabbitMq中消息可靠性的示例分析”這篇文章的所有內容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內容對大家有所幫助，如果還想學習更多知識，歡迎關注億速云行業資訊頻道！