2. 程式人生 > >RabbitMQ的持久化機制

RabbitMQ的持久化機制

阿新 • • 發佈：2019-01-06

一.問題的引出 
RabbitMQ的一大特色是訊息的可靠性,那麼它是如何保證訊息可靠性的呢?——訊息持久化。為了保證RabbitMQ在退出,服務重啟或者crash等異常情況下,也不會丟失訊息，我們可以將Queue,Exchange,Message都設定為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ訊息不會丟失。當然還是會有一些小概率事件會導致訊息丟失。

二.Queue的持久化 
1.檢視存在的佇列和訊息數量 
在windows環境下，在rabbitmq的安裝目錄/sbin下，通過rabbitmqctl.bat list_queues檢視 
這裡寫圖片描述

 
這邊啟動了兩個producer,分別生成兩個佇列hello 和 hello1,並且他們都有一個訊息存在 
重啟RabbitMQ Server，模擬故障 
這裡寫圖片描述 
可以看到重啟後兩個佇列都消失了. 
2.持久化佇列 
Queue的持久化是通過durable=true來實現的。 
一般程式中這麼使用：

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
//關鍵的是第二個引數設定為true,即durable=true. channel.queueDeclare("queue.persistent.name", true, false, false, null);
  • 1
  • 2
  • 3
  • 4

Channel類中queueDeclare的完整定義如下：

 /**
     * Declare a queue
     * @see com.rabbitmq.client.AMQP.Queue.Declare
     * @see com.rabbitmq.client.AMQP.Queue.DeclareOk
     * @param queue the name of the queue
     * @param durable true if we are declaring a durable queue (the queue will survive a server restart) * @param exclusive true if we are declaring an exclusive queue (restricted to this connection) * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use) * @param arguments other properties (construction arguments) for the queue * @return a declaration-confirm method to indicate the queue was successfully declared * @throws java.io.IOException if an error is encountered */ Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

引數說明：

queue：queue的名稱

exclusive：排他佇列，如果一個佇列被宣告為排他佇列，該佇列僅對首次申明它的連線可見，並在連線斷開時自動刪除。這裡需要注意三點：1. 排他佇列是基於連線可見的，同一連線的不同通道是可以同時訪問同一連線建立的排他佇列；2.“首次”，如果一個連線已經聲明瞭一個排他佇列，其他連線是不允許建立同名的排他佇列的，這個與普通佇列不同；3.即使該佇列是持久化的，一旦連線關閉或者客戶端退出，該排他佇列都會被自動刪除的，這種佇列適用於一個客戶端傳送讀取訊息的應用場景。

autoDelete：自動刪除，如果該佇列沒有任何訂閱的消費者的話，該佇列會被自動刪除。這種佇列適用於臨時佇列。

queueDeclare相關的有4種方法，分別是：

Queue.DeclareOk queueDeclare() throws IOException;
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException; void queueDeclareNoWait(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException; Queue.DeclareOk queueDeclarePassive(String queue) throws IOException;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

其中需要說明的是queueDeclarePassive(String queue)可以用來檢測一個queue是否已經存在。如果該佇列存在，則會返回true；如果不存在，就會返回異常，但是不會建立新的佇列。

我們就hello佇列持久化,在宣告佇列名稱時，持久化佇列，生產端和消費端都要. 
我們重複上面的操作，但是給hello佇列做持久化，而hello1不做，並重啟rabbitmq. 
這裡寫圖片描述 
可以看到重啟後，hello佇列還在，hello1佇列消失了，但是原本hello中的一條訊息也沒有儲存下來。所以在這邊我們僅僅做到了訊息佇列的持久化，還沒有做訊息持久化。

三.Message的持久化 
如果將Queue的持久化標識durable設定為true,則代表是一個持久的佇列，那麼在服務重啟之後，也會存在，因為服務會把持久化的queue存放在硬碟上，當服務重啟的時候，會重新載入之前被持久化的queue。佇列是可以被持久化，但是裡面的訊息是否為持久化那還要看訊息的持久化設定。也就是說，重啟之前那個Queue裡面還有沒發出去的訊息的話，重啟之後那佇列裡面是不是還存在原來的訊息，這個就要取決於傳送者在傳送訊息時對訊息的設定了。 
如果要在重啟後保持訊息的持久化必須設定訊息是持久化的標識。

channel.basicPublish("exchange.persistent", "persistent", MessageProperties.PERSISTENT_TEXT_PLAIN, "persistent_test_message".getBytes());
  • 1

這裡的關鍵是：MessageProperties.PERSISTENT_TEXT_PLAIN 
首先看一下basicPublish的方法：

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
void basicPublish(String exchange, String routingKey, boolean mandatory, BasicProperties props, byte[] body) throws IOException; void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException;
  • 1
  • 2
  • 3
  • 4
  • 5

exchange表示exchange的名稱 
routingKey表示routingKey的名稱 
body代表傳送的訊息體 
有關mandatory和immediate的詳細解釋可以參考：RabbitMQ之mandatory和immediate 
這裡關鍵的是BasicProperties props這個引數了，這裡看下BasicProperties的定義：

public BasicProperties(
            String contentType,//訊息型別如：text/plain
            String contentEncoding,//編碼 Map<String,Object> headers, //這裡的deliveryMode=1代表不持久化，deliveryMode=2代表持久化。 Integer deliveryMode,//1:nonpersistent 2:persistent Integer priority,//優先順序 String correlationId, String replyTo,//反饋佇列 String expiration,//expiration到期時間 String messageId, Date timestamp, String type, String userId, String appId, String clusterId)
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

上面的實現程式碼使用的是MessageProperties.PERSISTENT_TEXT_PLAIN，那麼這個又是什麼呢？

public static final BasicProperties PERSISTENT_TEXT_PLAIN =
    new BasicProperties("text/plain", null, null, 2, 0, null, null, null, null, null, null, null, null, null);
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

可以看到這其實就是講deliveryMode設定為2的BasicProperties的物件，為了方便程式設計而出現的一個東東。 換一種實現方式：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder(); builder.deliveryMode(2); AMQP.BasicProperties properties = builder.build(); channel.basicPublish("exchange.persistent", "persistent",properties, "persistent_test_message".getBytes());
  • 1
  • 2
  • 3
  • 4

設定了佇列和訊息的持久化之後，當broker服務重啟的之後，訊息依舊存在。單隻設定佇列持久化，重啟之後訊息會丟失；單隻設定訊息的持久化，重啟之後佇列消失，既而訊息也丟失。單單設定訊息持久化而不設定佇列的持久化顯得毫無意義。 
再以上面例子,生產端生成一個訊息，並重啟rabbitmq. 
這裡寫圖片描述 
可以看到，經過佇列和訊息持久化後的hello, 在重啟的情況下，佇列和訊息都存在，沒有消失,消費端再重啟後也是能正常接收的.

四.Exchange的持久化 
上面闡述了佇列的持久化和訊息的持久化，如果不設定exchange的持久化對訊息的可靠性來說沒有什麼影響，但是同樣如果exchange不設定持久化，那麼當broker服務重啟之後，exchange將不復存在，那麼既而傳送方rabbitmq producer就無法正常傳送訊息。這裡博主建議，同樣設定exchange的持久化。exchange的持久化設定也特別簡單，方法如下：

Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException;
Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments) throws IOException; Exchange.DeclareOk exchangeDeclare(String exchange, String type) throws IOException; Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException; void exchangeDeclareNoWait(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException; Exchange.DeclareOk exchangeDeclarePassive(String name) throws IOException;
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

一般只需要：channel.exchangeDeclare(exchangeName, “direct/topic/header/fanout”, true);即在宣告的時候講durable欄位設定為true即可。

五.關於Message的持久化的更進一步探討 
1.訊息什麼時候需要持久化？ 
根據 官方博文(http://www.rabbitmq.com/blog/2011/01/20/rabbitmq-backing-stores-databases-and-disks/) 的介紹，RabbitMQ在兩種情況下會將訊息寫入磁碟： 
- 訊息本身在publish的時候就要求訊息寫入磁碟 
- 記憶體緊張，需要將部分記憶體中的訊息轉移到磁碟

2.訊息什麼時候會刷到磁碟？ 
- 寫入檔案前會有一個Buffer，大小為1M（1048576），資料在寫入檔案時，首先會寫入到這個Buffer，如果Buffer已滿，則會將Buffer寫入到檔案（未必刷到磁碟） 
- 有個固定的刷盤時間：25ms，也就是不管Buffer滿不滿，每隔25ms，Buffer裡的資料及未重新整理到磁碟的檔案內容必定會刷到磁碟 
- 每次訊息寫入後，如果沒有後續寫入請求，則會直接將已寫入的訊息刷到磁碟：使用Erlang的receive x after 0來實現，只要程序的信箱裡沒有訊息，則產生一個timeout訊息，而timeout會觸發刷盤操作

3.訊息在磁碟檔案中的格式 
訊息保存於$MNESIA/msg_store_persistent/x.rdq檔案中，其中x為數字編號，從0開始，每個檔案最大為16M（16777216），超過這個大小會生成新的檔案，檔案編號加1。訊息以以下格式存在於檔案中：

<<Size:64, MsgId:16/binary, MsgBody>>

MsgId為RabbitMQ通過rabbit_guid:gen()每一個訊息生成的GUID，MsgBody會包含訊息對應的exchange，routing_keys，訊息的內容，訊息對應的協議版本，訊息內容格式（二進位制還是其它）等等。

4.檔案何時刪除？ 
當所有檔案中的垃圾訊息（已經被刪除的訊息）比例大於閾值（GARBAGE_FRACTION = 0.5）時，會觸發檔案合併操作（至少有三個檔案存在的情況下），以提高磁碟利用率。 
publish訊息時寫入內容，ack訊息時刪除內容（更新該檔案的有用資料大小），當一個檔案的有用資料等於0時，刪除該檔案。

 

5.將queue，exchange, message等都設定了持久化之後就能保證100%保證資料不丟失了嚒？ 
答案是否定的。 
首先，從consumer端來說，如果這時autoAck=true，那麼當consumer接收到相關訊息之後，還沒來得及處理就crash掉了，那麼這樣也算資料丟失，這種情況也好處理，只需將autoAck設定為false(方法定義如下)，然後在正確處理完訊息之後進行手動ack（channel.basicAck）.

String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
  • 1

其次，關鍵的問題是訊息在正確存入RabbitMQ之後，還需要有一段時間（這個時間很短，但不可忽視）才能存入磁碟之中，RabbitMQ並不是為每條訊息都做fsync的處理，可能僅僅儲存到cache中而不是物理磁碟上，在這段時間內RabbitMQ broker發生crash, 訊息儲存到cache但是還沒來得及落盤，那麼這些訊息將會丟失。那麼這個怎麼解決呢？首先可以引入RabbitMQ的mirrored-queue即映象佇列，這個相當於配置了副本，當master在此特殊時間內crash掉，可以自動切換到slave，這樣有效的保障了HA, 除非整個叢集都掛掉，這樣也不能完全的100%保障RabbitMQ不丟訊息，但比沒有mirrored-queue的要好很多，很多現實生產環境下都是配置了mirrored-queue的。還有要在producer引入事務機制或者Confirm機制來確保訊息已經正確的傳送至broker端，有關RabbitMQ的事務機制或者Confirm機制可以參考：RabbitMQ之訊息確認機制（事務+Confirm）. 幸虧本文的主題是討論RabbitMQ的持久化而不是可靠性，不然就一發不可收拾了。RabbitMQ的可靠性涉及producer端的確認機制、broker端的映象佇列的配置以及consumer端的確認機制，要想確保訊息的可靠性越高，那麼效能也會隨之而降，魚和熊掌不可兼得，關鍵在於選擇和取捨。

相關參考連結:http://jzhihui.iteye.com/blog/1642324 
訊息中介軟體收錄集:https://blog.csdn.net/u013256816/article/details/54743481

    1. RabbitMQ訊息佇列（三）：任務分發機制
    2. RabbitMQ之mandatory和immediate
    3. RabbitMQ之訊息確認機制（事務+Confirm）
    4. RabbitMQ持久化機制

相關推薦

RabbitMQ持久化機制

之前其實已經寫過一篇關於RabbitMQ持久化的 文章 ，但那篇文章側重程式碼層面的寫入流程，對於持久化操作何時發生以及什麼時候會重新整理到磁碟等問題其實都沒有搞清楚，這篇文章著重於關注這些問題。 訊息什麼時候需要持久化？ 根據 官方博文 的介紹，RabbitMQ在兩

RabbitMQ的持久化機制

一.問題的引出 RabbitMQ的一大特色是訊息的可靠性,那麼它是如何保證訊息可靠性的呢?——訊息持久化。為了保證RabbitMQ在退出,服務重啟或者crash等異常情況下,也不會丟失訊息，我們可以將Queue,Exchange,Message都設定為可持久化的（durable），這樣可以保證絕大部

細說Redis持久化機制

pac 文件太大 rdb redis持久化 三種方式 server tracking 場景 自己的 概述 Redis不僅能夠作為緩存來使用，也能夠作為內存數據庫。Redis作為內存數據庫使用時。必需要解決一個問題：數據的持久性。有些將Redis作為緩存使用的場

redis高級命令4 持久化機制 、事務

丟失 1-1 數據 生成 開啟 緩存 字符 set 我們 redis的事務是支持很簡單，基本沒有啥用我們來看下面的列子 我們開啟一個事務，在事務中執行了age 加1，set a4 ，還有對一個字符串進行加一，對字符串加1導致了事務失敗，按道理incr age在事務操作

redis的持久化機制

執行 末尾 redis服務器 請求 阻塞 aof文件 就是 而不是 自動 redis簡介 　　redis是一個基於內存的nosql數據庫，和傳統關系數據最大的區別就是數據是存在內存中而不是硬盤上，從而帶來了tps的巨大提升。但是基於內存的數據庫最大的缺陷就是機器斷電

Redis的持久化機制包括RBD和AOF兩種，對於這兩種持久化方式各有優勢

plain 同步數據 pen toc 默認 ocl 好的 dfs 操作系統 RDB機制的策略 RDB持久化是指在指定的時間間隔內將內存中的數據和操作通過快照的方式保存到redis bin目錄下的一個默認名為 dump.rdb的文件，可以通過配置設置自動的快照持久化的

redis持久化機制

toolbar 我們 等等 HERE 了解 OS row alt 數據 redis持久化機制 參考： https://www.cnblogs.com/xingzc/p/5988080.html https://blog.csdn.net/yujin2010good/

RabbitMQ死信機制實現延遲佇列

延遲佇列 延遲佇列儲存的物件肯定是對應的延時訊息，所謂”延時訊息”是指當訊息被髮送以後，並不想讓消費者立即拿到訊息，而是等待指定時間後，消費者才拿到這個訊息進行消費。 應用場景 三方支付，掃碼支付呼叫上游的掃碼介面，當掃碼有效期過後去呼叫查詢介面查詢結果。實現方式：每當一筆掃碼支付請求後

Redis學習之持久化機制

原文部落格地址: pjmike的部落格 前言 持久化就是將Redis記憶體中的資料寫入到磁碟中進行儲存，因為Redis是記憶體資料庫，資料都是存在記憶體中的，為了避免程序退出導致資料的丟失，所以需要將資料持久化到硬碟中，這樣下次Redis重啟後可以利用之前持久化的檔案實現資料恢復。 一般有兩種

Atitit 持久化 Persistence概念的藝術 目錄 1. 持久化是將程式資料在持久狀態和瞬時狀態間轉換的機制。 1 2. DBC就是一種持久化機制。檔案IO也是一種持久化機制。 2 3.

Atitit 持久化 Persistence概念的藝術   目錄 1. 持久化是將程式資料在持久狀態和瞬時狀態間轉換的機制。 1 2. DBC就是一種持久化機制。檔案IO也是一種持久化機制。 2 3. 日常持久化的方法 2 4. 理解與分類 3 4.1

redis之五 持久化機制(AOF和RDB)

redis中文官網：redis持久化 redis支援兩種持久化方式：RDB方式和AOF方式 一、RDB方式 在預設情況下，Redis將資料庫快照儲存在名字為dump.rdb的二進位制檔案中。你可以對Redis進行設定，讓它在“N秒內資料集至少有M個改動”這一條件被滿足時，自動

redis的持久化機制 (RDB&AOF)

為了能夠重用Redis資料，或者防止系統故障，我們需要將Redis中的資料寫入到磁碟空間中，即持久化。Redis提供了兩種不同的持久化方法可以將資料儲存在磁碟中，一種叫快照RDB，另一種叫只追加檔案AOF RDB 是什麼   在指定的時間間隔內將記憶體中的資料集快照寫入磁碟(Snapshot)，它恢復時

Redis提供的持久化機制（RDB和AOF）

來源：https://www.cnblogs.com/xingzc/p/5988080.html 　　Redis是一種面向“key-value”型別資料的分散式NoSQL資料庫系統，具有高效能、持久儲存、適應高併發應用場景等優勢。它雖然起步較晚，但發展卻十分迅速。  近日，Redis的

圖解分析redis的RDB和AOF這兩種持久化機制的工作原理

▌大綱 RDB和AOF兩種持久化機制的介紹 RDB持久化機制的優點 RDB持久化機制的缺點 AOF持久化機制的優點 AOF持久化機制的缺點 RDB和AOF到底該如何選擇   ▌RDB和AOF兩種持久化機制的介紹 1、

Spring Boot + RabbitMQ死信機制實現延遲佇列

本文轉載自部落格：http://blog.csdn.net/nexttake/article/details/78607486 ------------------------------------------------------------------------------------

RabbitMq-持久化

自動 而且 del pre text param pla truct rop RabbitMq-消息持久化 問題：怎樣保證消息不因生產者gg而丟失我們知道了如何在消費者的角度保證消息不丟失，但如果生產者gg了呢，消息同樣會丟失，生產者gg後會默認丟棄所有的消息，除非告訴它某

Redis配置AOF持久化機制

AOF持久化的配置 AOF持久化，預設是關閉的，預設是開啟RDB持久化 appendonly yes，可以開啟AOF持久化機制，在生產環境裡面，一般來說AOF都是要開啟的，除非你說隨便丟個幾分鐘的資料也無所謂 開啟AOF持久化機制之後，redis每次接收到一條寫命令，就會寫入日誌檔案

Redis配置RDB持久化機制

如何配置RDB持久化機制 (1) redis.conf檔案，也就是/etc/redis/6379.conf，去配置持久化 (2) save 60 1000 (每隔60s，如果有超過1000個key發生了變更，那麼就生成一個新的dump.rdb檔案，就是當前redis記憶體中完整的資料快

Redis的持久化機制-RDB和AOF

1.什麼是持久化         簡單的說就是指資料放到斷電後，仍然不會丟失的裝置上，也就是我們通常理解的硬碟上 2.資料庫在進行寫操作的時候，經歷那些過程 客戶端向資料庫服務端傳送寫操作（資料在客戶端的記憶體中） 資料庫服務端接收到寫

ActiveMQ的幾種訊息持久化機制

為了避免意外宕機以後丟失資訊，需要做到重啟後可以恢復訊息佇列，訊息系統一般都會採用持久化機制。 ActiveMQ的訊息持久化機制有JDBC，AMQ，KahaDB和LevelDB四種方式，無論使用哪種持久化方式，訊息的儲存邏輯都是一致的。 就是在傳送者將訊息傳送出