RabbitMQ之消息持久化(轉)

原文地址 https://blog.csdn.net/u013256816/article/details/60875666/

 

消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保證消息可靠性的呢——消息持久化。
為了保證RabbitMQ在退出或者crash等異常情況下數據沒有丟失，需要將queue，exchange和Message都持久化。

queue的持久化
queue的持久化是通過durable=true來實現的。
一般程序中這么使用：

Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("queue.persistent.name", true, false, false, null);

 

關鍵的是第二個參數設置為true,即durable=true.

　　

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;

　

參數說明：

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;

　　

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

消息的持久化
如過將queue的持久化標識durable設置為true,則代表是一個持久的隊列，那么在服務重啟之后，也會存在，因為服務會把持久化的queue存放在硬盤上，當服務重啟的時候，會重新什么之前被持久化的queue。隊列是可以被持久化，但是里面的消息是否為持久化那還要看消息的持久化設置。也就是說，重啟之前那個queue里面還沒有發出去的消息的話，重啟之后那隊列里面是不是還存在原來的消息，這個就要取決於發生着在發送消息時對消息的設置了。
如果要在重啟后保持消息的持久化必須設置消息是持久化的標識。

設置消息的持久化：

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

 

這里的關鍵是：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;

 

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,
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)

　　

這里的deliveryMode=1代表不持久化，deliveryMode=2代表持久化。

上面的實現代碼使用的是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);

　　

可以看到這其實就是講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());

 

設置了隊列和消息的持久化之后，當broker服務重啟的之后，消息依舊存在。單只設置隊列持久化，重啟之后消息會丟失；單只設置消息的持久化，重啟之后隊列消失，既而消息也丟失。單單設置消息持久化而不設置隊列的持久化顯得毫無意義。

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;

　　

一般只需要：channel.exchangeDeclare(exchangeName, “direct/topic/header/fanout”, true);即在聲明的時候講durable字段設置為true即可。

進一步討論
1.將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端的確認機制，要想確保消息的可靠性越高，那么性能也會隨之而降，魚和熊掌不可兼得，關鍵在於選擇和取舍。

2.消息什么時候刷到磁盤？
寫入文件前會有一個Buffer,大小為1M,數據在寫入文件時，首先會寫入到這個Buffer，如果Buffer已滿，則會將Buffer寫入到文件（未必刷到磁盤）。
有個固定的刷盤時間：25ms,也就是不管Buffer滿不滿，每個25ms，Buffer里的數據及未刷新到磁盤的文件內容必定會刷到磁盤。
每次消息寫入后，如果沒有后續寫入請求，則會直接將已寫入的消息刷到磁盤：使用Erlang的receive x after 0實現，只要進程的信箱里沒有消息，則產生一個timeout消息，而timeout會觸發刷盤操作。

欲了解更多消息中間件的內容，可以關注：消息中間件收錄集

參考資料
RabbitMQ消息隊列（三）：任務分發機制
RabbitMQ之mandatory和immediate
RabbitMQ之消息確認機制（事務+Confirm）
RabbitMQ持久化機制