RabbitMQ 發布訂閱持久化及持久化方式

RabbitMQ是一種重要的消息隊列中間件，在生產環境中，穩定是第一考慮。RabbitMQ廠家也深知開發者的聲音，穩定、可靠是第一考慮，為了消息傳輸的可靠性傳輸，RabbitMQ提供了多種途徑的消息持久化保證：Exchange持久化、Queue持久化及Message的持久化等。以保證RabbitMQ在重啟或Crash等異常情況下，消息不會丟失。RabbitMQ提供了簡單的參數配置來實現持久化操作。

簡單說明一下各種持久化方式：(描述代碼采用的是RabbitMQ.Client  SDK,  C#代碼)

Queue持久化：隊列是我們使用RabbitMQ進行數據傳輸的最多使用的方式，是進行點對點消息傳遞使用最多的方式。隊列的持久化是通過durable=true 來實現。

var connFactory = new ConnectionFactory();
Conn = connFactory.CreateConnection();
Model = Conn.CreateModel();
Model.QueueDeclare(q, false, false, false, null);　　

 

其中，QueueDeclare的定義：

/// <summary>(Spec method) Declare a queue.</summary>
        [AmqpMethodDoNotImplement(null)]
        QueueDeclareOk QueueDeclare(string queue, bool durable, bool exclusive,
            bool autoDelete, IDictionary<string, object> arguments);　　

 

參數說明：queue:隊列名稱。durable：設置是否執行持久化。如果設置為true，即durable=true，持久化實現的重要參數。

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

autoDelete:是否自動刪除。如果該隊列沒有任何訂閱的消費者的話，該隊列會被自動刪除。這種隊列適用於發布訂閱方式創建的臨時隊列。

 

消息的持久化：如果將一個隊列設置為持久化，那么會創建一個持久化的隊列，但並不意味着隊列中的消息也會持久化存儲。因此如果要保證消息在RabbitMQ出現異常時不會丟失，需要設定消息的持久化。

簡要說明一下消息持久化和隊列持久化的聯系：

隊列設置為持久化，那么在RabbitMQ重啟之后，持久化的隊列也會存在，並會保持和重啟前一致的隊列參數。

消息設置為持久化，在RabbitMQ重啟之后，持久化的消息也會存在。

那么就會出現一些矛盾的地方：

1、因為消息必須依附於隊列存在才有意義，那么如果隊列設置為非持久化，而消息設置為持久化。在RabbitMQ重啟之后，持久化的消息是否還存在呢？因為非持久化的隊列可能並不存在。

2、如果設置消息持久化為true，但隊列設置成排他性隊列，那么在RabbitMQ重啟之后，消息是否仍然存在。請自行查找分析，下次分析該問題。

             var sf = new ConnectionFactory();
            using (IConnection conn = cf.CreateConnection())
            {
                IModel ch = conn.CreateModel();

                   Model = Conn.CreateModel();
                   Model.QueueDeclare(queueName, true, false, false, null);　

string message = "Hello C# SSL Client World"; 11 byte[] msgBytes = System.Text.Encoding.UTF8.GetBytes(message);
                   //發送消息
                ch.BasicPublish("", queueName, null, msgBytes);

                bool noAck = false;
                BasicGetResult result = ch.BasicGet(qName, noAck);
                byte[] body = result.Body;
                string resultMessage = System.Text.Encoding.UTF8.GetString(body);

                Assert.AreEqual(message, resultMessage);
            }

通過RabbitMQ SDK發送消息至MQ非常簡單，通過BasicPublish即可。

 BasicPublish 的定義：

  /// <summary>
        /// (Spec method) Convenience overload of BasicPublish.
        /// </summary>
        /// <remarks>
        /// The publication occurs with mandatory=false
        /// </remarks>
        [AmqpMethodDoNotImplement(null)]
        void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, byte[] body);

設置消息持久化，需要設置basicProperties的DeliveryMode=2 (Non-persistent (1) or persistent (2)).

設置了隊列和消息持久化后，當服務重啟之后，消息仍然存在。只設置隊列持久化，不設置消息持久化，重啟之后消息會丟失；只設置消息持久化，不設置隊列持久化，在服務重啟后，隊列會消失，從而依附於隊列的消息也會丟失。只設置消息持久化而不設置隊列的持久化，毫無意義。

 

Exchange持久化：

為了實現一對多的消息發送，我們一般會采用發布訂閱模式，通過一個發送端、多個訂閱端來實現消息的分發。

發布訂閱模式存在一些問題：

1、如果消費者由於網絡或其他原因，與RabbitMQ的連接斷開，那么RabbitMQ會自動將與其對應的隊列刪除，當消息程序重新連接以后，無法獲取斷開前未來得及消費的消息。

2、如果RabbitMQ出現故障或Crash，那么在RabbitMQ  服務重啟之后，消費端未及時消費的消息也會丟失，並且如果Exchange 不設置成持久化，那么在MQ服務重啟之后，Exchange也不會存在。

  /// <summary>(Spec method) Declare an exchange.</summary>
        /// <remarks>
        /// The exchange is declared non-passive and non-internal.
        /// The "nowait" option is not exercised.
        /// </remarks>
        [AmqpMethodDoNotImplement(null)]
        void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete,
            IDictionary<string, object> arguments);

參數說明：exchange：RabbitMQ中定義的Exchange名稱，type：類型，包含fanout、topic、direct、headers，durable：持久化設置。設置成true，就可以設定exchange持久化存儲，autodelete：是否自動刪除。

exchange是實現發布訂閱的基礎，其類型包含fanout、headers、direct、、topic。我們本次僅討論類型為topic。

發布訂閱模式執行消息發送的流程：

 

總結：

RabbitMQ要實現發布訂閱持久化，按照消息的傳輸流程，可以分成三類：

Exchange 持久化：如果不設定Exchange持久化，那么在RabbitMQ由於某些異常等原因重啟之后，Exchange會丟失。Exchange丟失， 會影響發送端發送消息到RabbitMQ。

Queue持久化：發送端將消息發送至Exchange，Exchange將消息轉發至關聯的Queue。如果Queue不設置持久化，那么在RabbitMQ重啟之后，Queue信息會丟失。導致消息發送至Exchange，但Exchange不知道需要將該消息發送至哪些具體的隊列。

Message持久化：發送端將消息發送至Exchange，Exchange將消息轉發至關聯的Queue，消息存儲於具體的Queue中。如果RabbitMQ重啟之后，由於Message未設置持久化，那么消息會在重啟之后丟失。

為了保證發布訂閱的持久化，必須設置Exchange、Queue、Message的持久化，才可以保證消息最終不會丟失。

雖然持久化會造成性能損耗，但為了生產環境的數據一致性，這是我們必須做出的選擇。但我們可以通過設置消息過期時間、降低發送消息大小等其他方式來盡可能的降低MQ性能的降低。

擴展閱讀：

1、Exchange type：topic、fanout、direct、headers的不同。

2、消息的確認機制。

3、將Exchange、Queue、Message都設置持久化，能保證消息100%會被成功消費嗎？

答案肯定是否，天下沒有絕對的事情，尤其是復雜的MQ。

原因簡單介紹，一、如果消息的自動確認為true，那么在消息被接收以后，RabbitMQ就會刪除該消息，假如消費端此時宕機，那么消息就會丟失。因此需要將消息設置為手動確認。

二、設置手動確認會出現另一個問題，如果消息已被成功處理，但在消息確認過程中出現問題，那么在消費端重啟后，消息會重新被消費。

三、發送端為了保證消息會成功投遞，一般會設定重試。如果消息發送至RabbitMQ之后，在RabbitMQ回復已成功接收消息過程中出現異常，那么發送端會重新發送該消息，從而造成消息重復發送。

四、RabbitMQ的消息磁盤寫入，如果出現問題，也會造成消息丟失。

五、。。。。。

 

下期熱點問題：

1、Exchange type的不同

2、消息的確認與拒絕機制

3、優先級機制

 

RabbitMQ發布訂閱持久化方式：Exchange、Queue、Message持久化，隊列設定手動確認、AutoDelete=false。可以最大程度的保證消息不丟失。

附RabbitMQ發布訂閱持久化具體實現方式，參考代碼：

 

MQ SDK新增接口：
IMQSession新增方法：
/// <summary>
        /// 創建消息消費者
        /// </summary>
        /// <param name="topicName">主題名稱</param> 
        /// <param name="customTopicQueueName">自定義Topic關聯隊列名稱</param>
        /// <param name="isPersistence">是否持久化</param>
        /// <returns>消息消費者</returns>
        IMessageConsumer CreateTopicConsumer(string topicName, string customTopicQueueName, bool isPersistence = false);
調用方式：消費端需要明確指定需要消費的發布訂閱關聯隊列。例如配置中心熱部署，每個配置中心實例都需要指定唯一的關聯隊列名。
這樣就可以和正常的MAC隊列消費一樣，消費指定隊列消息。

實現方式，四個步驟：
1.創建持久化Topic(即持久化Exchange)：
  var service = MQServiceProvider.GetDefaultMQService();
            var messageText = "abc";
            ///創建Topic
            using (var connection = service.CreateConnection())
            {
                var session = connection.CreateSession(MessageAckMode.IndividualAcknowledge);
                var messageCreator = service.GetMessageCreator();
                var message = messageCreator.CreateMessage(messageText);
                message.IsPersistent = true;
                var producer = session.CreateProducer();
                var topic = session.DeclareTopic(topicName, true);
            }
2.定義消費者Consumer：
List<string> queueList = new List<string>() {
                "guozhiqi1",
                "guozhiqi2",
                "guozhiqi3",
                "guozhiqi4",
                "guozhiqi5",
                "guozhiqi6",
                "guozhiqi7",
                "guozhiqi8",
                "guozhiqi9",
            };
            //var service = MQServiceProvider.GetDefaultMQService();
            //var messageText = "abc" + DateTime.Now.ToShortTimeString();
            //定義消費者
            using (var connection1 = service.CreateConnection())
            {
                var session1 = connection1.CreateSession(MessageAckMode.IndividualAcknowledge);
                foreach (var item in queueList)
                {
                    session1.DeclareQueue(item, true);
                    var consumer = session1.CreateTopicConsumer(topicName, item, true);
                }
            }
3.發送消息到Topic
 //發送消息
            for (int i = 0; i <= 100; i++)
            {
                using (var connection = service.CreateConnection())
                {
                    var session = connection.CreateSession(MessageAckMode.IndividualAcknowledge);
                    var messageCreator = service.GetMessageCreator();
                    var message = messageCreator.CreateMessage(messageText);
                    message.IsPersistent = true;//設置持久化
                   message.TimeToLive = TimeSpan.FromSeconds(30);//設置過期時間
                    var producer = session.CreateProducer();
                    var topic = session.DeclareTopic(topicName, true);
                    producer.Send(message, topic);
                }
            }
4.從隊列接收消息
Parallel.ForEach(queueList, (item) =>
            {
                while (true)
                {
                    //接收消息
                    using (var connection1 = service.CreateConnection())
                    {
                        var session1 = connection1.CreateSession(MessageAckMode.IndividualAcknowledge);

                        session1.DeclareQueue(item, true);
                        var consumer = session1.CreateTopicConsumer(topicName, item, true);
                        var topic = session1.DeclareTopic(topicName, true);
                        var receivedmessage = consumer.Receive(topic);
                        var textMessage = receivedmessage as ITextMessage;

                        Assert.AreEqual(messageText, textMessage.Body);
                        consumer.Acknowledge(receivedmessage);
                    }
                }

            });