有很多人問過我這么一類問題：RabbitMQ如何確保消息可靠？很多時(shí)候，筆者的回答都是：說來話長的事情何來長話短說。的確，要確保消息可靠不只是單單幾句就能夠敘述明白的，包括Kafka也是如此?？煽坎⒉皇且粋€(gè)絕對的概念，曾經(jīng)有人也留言說過類似全部磁盤損毀也會(huì)導(dǎo)致消息丟失，筆者戲答：還有機(jī)房被炸了也會(huì)導(dǎo)致消息丟失?？煽啃允且粋€(gè)相對的概念，在條件合理的范圍內(nèi)系統(tǒng)所能確保的多少個(gè)9的可靠性。一切盡可能的趨于完美而無法企及于完美。
我們可以盡可能的確保RabbitMQ的消息可靠。在詳細(xì)論述RabbitMQ的消息可靠性之前，我們先來回顧下消息在RabbitMQ中的經(jīng)由之路。
如圖所示，從AMQP協(xié)議層面上來說：
成都創(chuàng)新互聯(lián)公司專業(yè)為企業(yè)提供寧強(qiáng)網(wǎng)站建設(shè)、寧強(qiáng)做網(wǎng)站、寧強(qiáng)網(wǎng)站設(shè)計(jì)、寧強(qiáng)網(wǎng)站制作等企業(yè)網(wǎng)站建設(shè)、網(wǎng)頁設(shè)計(jì)與制作、寧強(qiáng)企業(yè)網(wǎng)站模板建站服務(wù)，10年寧強(qiáng)做網(wǎng)站經(jīng)驗(yàn)，不只是建網(wǎng)站，更提供有價(jià)值的思路和整體網(wǎng)絡(luò)服務(wù)。
消息先從生產(chǎn)者Producer出發(fā)到達(dá)交換器Exchange；
交換器Exchange根據(jù)路由規(guī)則將消息轉(zhuǎn)發(fā)對應(yīng)的隊(duì)列Queue之上；
消息在隊(duì)列Queue上進(jìn)行存儲(chǔ)；
消費(fèi)者Consumer訂閱隊(duì)列Queue并進(jìn)行消費(fèi)。
我們對于消息可靠性的分析也從這四個(gè)階段來一一探討。
消息從生產(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、channel.txCommit以及channel.txRollback。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)得到了處理。
事務(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方法（即Confirm.Select命令）將信道設(shè)置為confirm模式，之后RabbitMQ會(huì)返回 Confirm.Select-Ok命令表示同意生產(chǎn)者將當(dāng)前信道設(shè)置為confirm模式。所有被發(fā)送的后續(xù)消息都被ack或者nack一次，不會(huì)出現(xiàn)一條消息即被ack又被nack的情況。并且RabbitMQ也并沒有對消息被confirm的快慢做任何保證。
事務(wù)機(jī)制和publisher confirm機(jī)制兩者是互斥的，不能共存。如果企圖將已開啟事務(wù)模式的信道再設(shè)置為publisher confirm模式，RabbitMQ會(huì)報(bào)錯(cuò)：{amqp_error, precondition_failed, “cannot switch from tx to confirm mode”, ‘confirm.select’}，或者如果企圖將已開啟publisher confirm模式的信道在設(shè)置為事務(wù)模式的話，RabbitMQ也會(huì)報(bào)錯(cuò)：{amqp_error, precondition_failed, “cannot switch from confirm to tx mode”, ‘tx.select’ }。
事務(wù)機(jī)制和publisher confirm機(jī)制確保的是消息能夠正確的發(fā)送至RabbitMQ，這里的“發(fā)送至RabbitMQ”的含義是指消息被正確的發(fā)往至RabbitMQ的交換器，如果此交換器沒有匹配的隊(duì)列的話，那么消息也將會(huì)丟失。所以在使用這兩種機(jī)制的時(shí)候要確保所涉及的交換器能夠有匹配的隊(duì)列。更進(jìn)一步的講，發(fā)送方要配合mandatory參數(shù)或者備份交換器一起使用來提高消息傳輸?shù)目煽啃浴?/p>
mandatory和immediate是channel.basicPublish方法中的兩個(gè)參數(shù)，它們都有當(dāng)消息傳遞過程中不可達(dá)目的地時(shí)將消息返回給生產(chǎn)者的功能。而RabbitMQ提供的備份交換器（Alternate Exchange）可以將未能被交換器路由的消息（沒有綁定隊(duì)列或者沒有匹配的綁定）存儲(chǔ)起來，而不用返回給客戶端。
RabbitMQ 3.0版本開始去掉了對于immediate參數(shù)的支持，對此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)者沒有成功的將消息路由到隊(duì)列，此時(shí)RabbitMQ會(huì)通過Basic.Return返回“mandatory test”這條消息，之后生產(chǎn)者客戶端通過ReturnListener監(jiān)聽到了這個(gè)事件，上面代碼的最后輸出應(yīng)該是“Basic.Return返回的結(jié)果是：mandatory test”。
生產(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í)候去處理這些消息。 可以通過在聲明交換器（調(diào)用channel.exchangeDeclare方法）的時(shí)候添加alternate-exchange參數(shù)來實(shí)現(xiàn)，也可以通過策略的方式實(shí)現(xiàn)。如果兩者同時(shí)使用的話，前者的優(yōu)先級更高，會(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ì)列。
備份交換器其實(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ù)無效。
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è)置為持久化。通過將消息的投遞模式（BasicProperties中的deliveryMode屬性）設(shè)置為2即可實(shí)現(xiàn)消息的持久化。
設(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)（master）在此特殊時(shí)間內(nèi)掛掉，可以自動(dòng)切換到從節(jié)點(diǎn)（slave），這樣有效的保證了高可用性，除非整個(gè)集群都掛掉。雖然這樣也不能完全的保證RabbitMQ消息不丟失（比如機(jī)房被炸。。。），但是配置了鏡像隊(duì)列要比沒有配置鏡像隊(duì)列的可靠性要高很多，在實(shí)際生產(chǎn)環(huán)境中的關(guān)鍵業(yè)務(wù)隊(duì)列一般都會(huì)設(shè)置鏡像隊(duì)列。
進(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，對于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)之中。對于這種情況，筆者的建議是在出現(xiàn)無法正確消費(fèi)的消息時(shí)不要采用requeue的方式來確保消息可靠性，而是重新投遞到新的隊(duì)列中，比如設(shè)定的死信隊(duì)列中，以此可以避免前面所說的死循環(huán)而又可以確保相應(yīng)的消息不丟失。對于死信隊(duì)列中的消息可以用另外的方式來消費(fèi)分析，以便找出問題的根本。
本文的重點(diǎn)是你有沒有收獲與成長，其余的都不重要，希望讀者們能謹(jǐn)記這一點(diǎn)。同時(shí)我經(jīng)過多年的收藏目前也算收集到了一套完整的學(xué)習(xí)資料，包括但不限于：分布式架構(gòu)、高可擴(kuò)展、高性能、高并發(fā)、Jvm性能調(diào)優(yōu)、Spring，MyBatis，Nginx源碼分析，redis，ActiveMQ、、Mycat、Netty、Kafka、MySQL、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多個(gè)知識(shí)點(diǎn)高級進(jìn)階干貨，希望對想成為架構(gòu)師的朋友有一定的參考和幫助
