概述

RabbitMQ如何確保消息可靠菇存？

這是一個(gè)很難回答的問題潜的，要確保消息可靠不只是單單幾句就能夠敘述明白的，包括Kafka也是如此永部《琅ⅲ可靠是一個(gè)相對(duì)的概念，比較客觀的衡量標(biāo)準(zhǔn)是在條件合理的范圍內(nèi)系統(tǒng)所能確保的幾個(gè)9的可靠性苔埋。

在詳細(xì)論述RabbitMQ的消息可靠性之前懦砂，我們先來回顧下消息在RabbitMQ中的經(jīng)由之路。

1.jpeg

如圖所示组橄，從AMQP協(xié)議層面上來說：

• 消息先從生產(chǎn)者Producer出發(fā)到達(dá)交換器Exchange荞膘；
• 交換器Exchange根據(jù)路由規(guī)則將消息轉(zhuǎn)發(fā)對(duì)應(yīng)的隊(duì)列Queue之上；
• 消息在隊(duì)列Queue上進(jìn)行存儲(chǔ)玉工；
• 消費(fèi)者Consumer訂閱隊(duì)列Queue并進(jìn)行消費(fèi)羽资。

我們對(duì)于消息可靠性的分析也從這四個(gè)階段來一一探討。

階段1

消息從生產(chǎn)者發(fā)出到達(dá)交換器Exchange遵班，在這個(gè)過程中可以發(fā)生各種情況屠升，生產(chǎn)者客戶端發(fā)送出去之后可以發(fā)生網(wǎng)絡(luò)丟包、網(wǎng)絡(luò)故障等造成消息丟失狭郑。一般情況下如果不采取措施腹暖，生產(chǎn)者無法感知消息是否已經(jīng)正確無誤的發(fā)送到交換器中。如果消息在傳輸?shù)紼xchange的過程中發(fā)生失敗而可以讓生產(chǎn)者感知的話翰萨，生產(chǎn)者可以進(jìn)行進(jìn)一步的處理動(dòng)作脏答，比如重新投遞相關(guān)消息以確保消息的可靠性。

為此AMQP協(xié)議在建立之初就考慮到這種情況而提供了事務(wù)機(jī)制缨历。RabbitMQ客戶端中與事務(wù)機(jī)制相關(guān)的方法有三個(gè)：

• channel.txSelect：用于將當(dāng)前的信道設(shè)置成事務(wù)模式
• channel.txCommit：用于提交事務(wù)
• channel.txRollback：用于事務(wù)回滾

在通過channel.txSelect方法開啟事務(wù)之后以蕴，我們便可以發(fā)布消息給RabbitMQ了，如果事務(wù)提交成功辛孵，則消息一定到達(dá)了RabbitMQ中丛肮，如果在事務(wù)提交執(zhí)行之前由于RabbitMQ異常崩潰或者其他原因拋出異常，這個(gè)時(shí)候我們便可以將其捕獲魄缚，進(jìn)而通過執(zhí)行channel.txRollback方法來實(shí)現(xiàn)事務(wù)回滾宝与。注意這里的RabbitMQ中的事務(wù)機(jī)制與大多數(shù)數(shù)據(jù)庫中的事務(wù)概念并不相同，需要注意區(qū)分冶匹。

事務(wù)確實(shí)能夠解決消息發(fā)送方和RabbitMQ之間消息確認(rèn)的問題习劫，只有消息成功被RabbitMQ接收，事務(wù)才能提交成功嚼隘，否則我們便可在捕獲異常之后進(jìn)行事務(wù)回滾诽里，與此同時(shí)可以進(jìn)行消息重發(fā)。但是使用事務(wù)機(jī)制的話會(huì)“吸干”RabbitMQ的性能飞蛹，那么有沒有更好的方法既能保證消息發(fā)送方確認(rèn)消息已經(jīng)正確送達(dá)谤狡，又能基本上不帶來性能上的損失呢灸眼？從AMQP協(xié)議層面來看并沒有更好的辦法，但是RabbitMQ提供了一個(gè)改進(jìn)方案墓懂，即發(fā)送方確認(rèn)機(jī)制（publisher confirm）焰宣。

生產(chǎn)者將信道設(shè)置成confirm（確認(rèn)）模式，一旦信道進(jìn)入confirm模式捕仔，所有在該信道上面發(fā)布的消息都會(huì)被指派一個(gè)唯一的ID（從1開始）匕积，一旦消息被投遞到所有匹配的隊(duì)列之后，RabbitMQ就會(huì)發(fā)送一個(gè)確認(rèn)（Basic.Ack）給生產(chǎn)者（包含消息的唯一ID）榜跌，這就使得生產(chǎn)者知曉消息已經(jīng)正確到達(dá)了目的地了闪唆。RabbitMQ回傳給生產(chǎn)者的確認(rèn)消息中的deliveryTag包含了確認(rèn)消息的序號(hào)，此外RabbitMQ也可以設(shè)置channel.basicAck方法中的multiple參數(shù)斜做，表示到這個(gè)序號(hào)之前的所有消息都已經(jīng)得到了處理苞氮。

2.jpeg

事務(wù)機(jī)制在一條消息發(fā)送之后會(huì)使發(fā)送端阻塞湾揽，以等待RabbitMQ的回應(yīng)瓤逼，之后才能繼續(xù)發(fā)送下一條消息。相比之下库物，發(fā)送方確認(rèn)機(jī)制最大的好處在于它是異步的霸旗，一旦發(fā)布一條消息，生產(chǎn)者應(yīng)用程序就可以在等信道返回確認(rèn)的同時(shí)繼續(xù)發(fā)送下一條消息戚揭，當(dāng)消息最終得到確認(rèn)之后诱告，生產(chǎn)者應(yīng)用便可以通過回調(diào)方法來處理該確認(rèn)消息，如果RabbitMQ因?yàn)樽陨韮?nèi)部錯(cuò)誤導(dǎo)致消息丟失民晒，就會(huì)發(fā)送一條nack（Basic.Nack）命令精居，生產(chǎn)者應(yīng)用程序同樣可以在回調(diào)方法中處理該nack命令。

生產(chǎn)者通過調(diào)用channel.confirmSelect方法將信道設(shè)置為confirm模式潜必，之后RabbitMQ會(huì)返回 Confirm.Select-Ok命令表示同意生產(chǎn)者將當(dāng)前信道設(shè)置為confirm模式靴姿。所有被發(fā)送的后續(xù)消息都被ack或者nack一次，不會(huì)出現(xiàn)一條消息即被ack又被nack的情況磁滚。并且RabbitMQ也并沒有對(duì)消息被confirm的快慢做任何保證佛吓。

事務(wù)機(jī)制和publisher confirm機(jī)制兩者是互斥的，不能共存垂攘。如果企圖將已開啟事務(wù)模式的信道再設(shè)置為publisher confirm模式，RabbitMQ會(huì)報(bào)錯(cuò)。

事務(wù)機(jī)制和publisher confirm機(jī)制確保的是消息能夠正確的發(fā)送至RabbitMQ砸逊，這里的“發(fā)送至RabbitMQ”的含義是指消息被正確的發(fā)往至RabbitMQ的交換器笼平，如果此交換器沒有匹配的隊(duì)列的話，那么消息也將會(huì)丟失陨仅。所以在使用這兩種機(jī)制的時(shí)候要確保所涉及的交換器能夠有匹配的隊(duì)列津滞。更進(jìn)一步的講耕陷，發(fā)送方要配合mandatory參數(shù)或者備份交換器一起使用來提高消息傳輸?shù)目煽啃浴?/p>

階段2

mandatory和immediate是channel.basicPublish方法中的兩個(gè)參數(shù)，它們都有當(dāng)消息傳遞過程中不可達(dá)目的地時(shí)將消息返回給生產(chǎn)者的功能据沈。而RabbitMQ提供的備份交換器（Alternate Exchange）可以將未能被交換器路由的消息（沒有綁定隊(duì)列或者沒有匹配的綁定）存儲(chǔ)起來哟沫，而不用返回給客戶端。

RabbitMQ 3.0版本開始去掉了對(duì)于immediate參數(shù)的支持锌介，對(duì)此RabbitMQ官方解釋是：immediate參數(shù)會(huì)影響鏡像隊(duì)列的性能嗜诀，增加代碼復(fù)雜性，建議采用TTL和DLX的方法替代孔祸。所以本文只簡單介紹mandatory和備份交換器隆敢。

當(dāng)mandatory參數(shù)設(shè)為true時(shí)，交換器無法根據(jù)自身的類型和路由鍵找到一個(gè)符合條件的隊(duì)列的話崔慧，那么RabbitMQ會(huì)調(diào)用Basic.Return命令將消息返回給生產(chǎn)者拂蝎。當(dāng)mandatory參數(shù)設(shè)置為false時(shí)，出現(xiàn)上述情形的話惶室，消息直接被丟棄温自。 那么生產(chǎn)者如何獲取到?jīng)]有被正確路由到合適隊(duì)列的消息呢？這時(shí)候可以通過調(diào)用channel.addReturnListener來添加ReturnListener監(jiān)聽器實(shí)現(xiàn)皇钞。使用mandatory參數(shù)的關(guān)鍵代碼如下所示：

channel.basicPublish(EXCHANGE_NAME, "", true, MessageProperties.PERSISTENT_TEXT_PLAIN, "mandatory test".getBytes());
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返回的結(jié)果是：" + message);
    }
});

生產(chǎn)者可以通過ReturnListener中返回的消息來重新投遞或者其它方案來提高消息的可靠性悼泌。

備份交換器，英文名稱Alternate Exchange夹界，簡稱AE馆里。生產(chǎn)者在發(fā)送消息的時(shí)候如果不設(shè)置mandatory參數(shù)，那么消息在未被路由的情況下將會(huì)丟失可柿，如果設(shè)置了mandatory參數(shù)鸠踪，那么需要添加ReturnListener的編程邏輯，生產(chǎn)者的代碼將變得復(fù)雜化复斥。如果你不想復(fù)雜化生產(chǎn)者的編程邏輯营密，又不想消息丟失，那么可以使用備份交換器永票，這樣可以將未被路由的消息存儲(chǔ)在RabbitMQ中卵贱，再在需要的時(shí)候去處理這些消息。 可以通過在聲明交換器的時(shí)候添加alternate-exchange參數(shù)來實(shí)現(xiàn)侣集，也可以通過策略的方式實(shí)現(xiàn)键俱。如果兩者同時(shí)使用的話，前者的優(yōu)先級(jí)更高世分，會(huì)覆蓋掉Policy的設(shè)置编振。

參考下圖，如果此時(shí)我們發(fā)送一條消息到normalExchange上，當(dāng)路由鍵等于“normalKey”的時(shí)候踪央，消息能正確路由到normalQueue這個(gè)隊(duì)列中臀玄。如果路由鍵設(shè)為其他值，比如“errorKey”畅蹂，即消息不能被正確的路由到與normalExchange綁定的任何隊(duì)列上健无，此時(shí)就會(huì)發(fā)送給myAe，進(jìn)而發(fā)送到unroutedQueue這個(gè)隊(duì)列液斜。

3.jpeg

備份交換器其實(shí)和普通的交換器沒有太大的區(qū)別累贤，為了方便使用，建議設(shè)置為fanout類型少漆，如若讀者想設(shè)置為direct或者topic的類型也沒有什么不妥臼膏。需要注意的是消息被重新發(fā)送到備份交換器時(shí)的路由鍵和從生產(chǎn)者發(fā)出的路由鍵是一樣的。備份交換器的實(shí)質(zhì)就是原有交換器的一個(gè)“備胎”示损，所有無法正確路由的消息都發(fā)往這個(gè)備份交換器中渗磅，可以為所有的交換器設(shè)置同一個(gè)AE，不過這里需要提前確保的是AE已經(jīng)正確的綁定了隊(duì)列检访，最好類型也是fanout的始鱼。

如果備份交換器和mandatory參數(shù)一起使用，那么mandatory參數(shù)無效烛谊。

階段3

mandatory或者AE可以讓消息在路由到隊(duì)列之前得到極大的可靠性保障风响，但是消息存入隊(duì)列之后的可靠性又如何保證嘉汰？

首先是持久化丹禀。持久化可以提高隊(duì)列的可靠性，以防在異常情況（重啟鞋怀、關(guān)閉双泪、宕機(jī)等）下的數(shù)據(jù)丟失。隊(duì)列的持久化是通過在聲明隊(duì)列時(shí)將durable參數(shù)置為true實(shí)現(xiàn)的密似，如果隊(duì)列不設(shè)置持久化焙矛，那么在RabbitMQ服務(wù)重啟之后，相關(guān)隊(duì)列的元數(shù)據(jù)將會(huì)丟失残腌，此時(shí)數(shù)據(jù)也會(huì)丟失村斟。正所謂“皮之不存，毛將焉附”抛猫，隊(duì)列都沒有了蟆盹，消息又能存在哪里呢？隊(duì)列的持久化能保證其本身的元數(shù)據(jù)不會(huì)因異常情況而丟失闺金，但是并不能保證內(nèi)部所存儲(chǔ)的消息不會(huì)丟失逾滥。要確保消息不會(huì)丟失，需要將其設(shè)置為持久化败匹。

設(shè)置了隊(duì)列和消息的持久化寨昙，當(dāng)RabbitMQ服務(wù)重啟之后讥巡，消息依舊存在。單單只設(shè)置隊(duì)列持久化舔哪，重啟之后消息會(huì)丟失欢顷；單單只設(shè)置消息的持久化，重啟之后隊(duì)列消失捉蚤，既而消息也丟失吱涉。單單設(shè)置消息持久化而不設(shè)置隊(duì)列的持久化顯得毫無意義。

在持久化的消息正確存入RabbitMQ之后外里，還需要有一段時(shí)間（雖然很短怎爵，但是不可忽視）才能存入磁盤之中。RabbitMQ并不會(huì)為每條消息都做同步存盤（調(diào)用內(nèi)核的fsync6方法）的處理盅蝗，可能僅僅保存到操作系統(tǒng)緩存之中而不是物理磁盤之中鳖链。如果在這段時(shí)間內(nèi)RabbitMQ服務(wù)節(jié)點(diǎn)發(fā)生了宕機(jī)、重啟等異常情況墩莫，消息保存還沒來得及落盤芙委，那么這些消息將會(huì)丟失。

如果在Phase1中采用了事務(wù)機(jī)制或者publisher confirm機(jī)制的話狂秦，服務(wù)端的返回是在消息落盤之后執(zhí)行的灌侣，這樣可以進(jìn)一步的提高了消息的可靠性。但是即便如此也無法避免單機(jī)故障且無法修復(fù)（比如磁盤損毀）而引起的消息丟失裂问，這里就需要引入鏡像隊(duì)列侧啼。鏡像隊(duì)列相當(dāng)于配置了副本，絕大多數(shù)分布式的東西都有多副本的概念來確保HA堪簿。在鏡像隊(duì)列中痊乾，如果主節(jié)點(diǎn)在此特殊時(shí)間內(nèi)掛掉，可以自動(dòng)切換到從節(jié)點(diǎn)（slave）椭更，這樣有效的保證了高可用性哪审，除非整個(gè)集群都掛掉。雖然這樣也不能完全的保證RabbitMQ消息不丟失虑瀑，但是配置了鏡像隊(duì)列要比沒有配置鏡像隊(duì)列的可靠性要高很多湿滓，在實(shí)際生產(chǎn)環(huán)境中的關(guān)鍵業(yè)務(wù)隊(duì)列一般都會(huì)設(shè)置鏡像隊(duì)列。

階段4

進(jìn)一步的從消費(fèi)者的角度來說舌狗，如果在消費(fèi)者接收到相關(guān)消息之后叽奥，還沒來得及處理就宕機(jī)了，這樣也算數(shù)據(jù)丟失把夸。

為了保證消息從隊(duì)列可靠地達(dá)到消費(fèi)者而线，RabbitMQ提供了消息確認(rèn)機(jī)制（message acknowledgement）。消費(fèi)者在訂閱隊(duì)列時(shí)，可以指定autoAck參數(shù)膀篮，當(dāng)autoAck等于false時(shí)嘹狞，RabbitMQ會(huì)等待消費(fèi)者顯式地回復(fù)確認(rèn)信號(hào)后才從內(nèi)存（或者磁盤）中移去消息（實(shí)質(zhì)上是先打上刪除標(biāo)記，之后再刪除）誓竿。當(dāng)autoAck等于true時(shí)磅网，RabbitMQ會(huì)自動(dòng)把發(fā)送出去的消息置為確認(rèn)，然后從內(nèi)存（或者磁盤）中刪除筷屡，而不管消費(fèi)者是否真正的消費(fèi)到了這些消息涧偷。

采用消息確認(rèn)機(jī)制后，只要設(shè)置autoAck參數(shù)為false毙死，消費(fèi)者就有足夠的時(shí)間處理消息（任務(wù)）燎潮，不用擔(dān)心處理消息過程中消費(fèi)者進(jìn)程掛掉后消息丟失的問題，因?yàn)镽abbitMQ會(huì)一直等待持有消息直到消費(fèi)者顯式調(diào)用Basic.Ack命令為止扼倘。

當(dāng)autoAck參數(shù)置為false确封，對(duì)于RabbitMQ服務(wù)端而言，隊(duì)列中的消息分成了兩個(gè)部分：一部分是等待投遞給消費(fèi)者的消息再菊；一部分是已經(jīng)投遞給消費(fèi)者爪喘，但是還沒有收到消費(fèi)者確認(rèn)信號(hào)的消息。如果RabbitMQ一直沒有收到消費(fèi)者的確認(rèn)信號(hào)纠拔，并且消費(fèi)此消息的消費(fèi)者已經(jīng)斷開連接秉剑，則RabbitMQ會(huì)安排該消息重新進(jìn)入隊(duì)列，等待投遞給下一個(gè)消費(fèi)者稠诲，當(dāng)然也有可能還是原來的那個(gè)消費(fèi)者侦鹏。

RabbitMQ不會(huì)為未確認(rèn)的消息設(shè)置過期時(shí)間，它判斷此消息是否需要重新投遞給消費(fèi)者的唯一依據(jù)是消費(fèi)該消息的消費(fèi)者連接是否已經(jīng)斷開吕粹，這么設(shè)計(jì)的原因是RabbitMQ允許消費(fèi)者消費(fèi)一條消息的時(shí)間可以很久很久种柑。

如果消息消費(fèi)失敗，也可以調(diào)用Basic.Reject或者Basic.Nack來拒絕當(dāng)前消息而不是確認(rèn)匹耕，如果只是簡單的拒絕那么消息會(huì)丟失，需要將相應(yīng)的requeue參數(shù)設(shè)置為true荠雕，那么RabbitMQ會(huì)重新將這條消息存入隊(duì)列稳其，以便可以發(fā)送給下一個(gè)訂閱的消費(fèi)者。如果requeue參數(shù)設(shè)置為false的話炸卑，RabbitMQ立即會(huì)把消息從隊(duì)列中移除既鞠，而不會(huì)把它發(fā)送給新的消費(fèi)者。

還有一種情況需要考慮：requeue的消息是存入隊(duì)列頭部的盖文，即可以快速的又被發(fā)送給消費(fèi)嘱蛋，如果此時(shí)消費(fèi)者又不能正確的消費(fèi)而又requeue的話就會(huì)進(jìn)入一個(gè)無盡的循環(huán)之中。對(duì)于這種情況，建議是在出現(xiàn)無法正確消費(fèi)的消息時(shí)不要采用requeue的方式來確保消息可靠性洒敏，而是重新投遞到新的隊(duì)列中龄恋，比如設(shè)定的死信隊(duì)列中，以此可以避免前面所說的死循環(huán)而又可以確保相應(yīng)的消息不丟失凶伙。對(duì)于死信隊(duì)列中的消息可以用另外的方式來消費(fèi)分析郭毕，以便找出問題的根本。