一、前情提示

上一篇文章《??MQ保證讀寫(xiě)消息不丟失，這個(gè)你都不會(huì)就等著被開(kāi)除吧...??》,我們初步介紹了之前制定的那些消息中間件數(shù)據(jù)不丟失的技術(shù)方案遺留的問(wèn)題。

一個(gè)最大的問(wèn)題，就是生產(chǎn)者投遞出去的消息，可能會(huì)丟失。

丟失的原因有很多，比如消息在網(wǎng)絡(luò)傳輸?shù)揭话氲臅r(shí)候因?yàn)榫W(wǎng)絡(luò)故障就丟了，或者是消息投遞到MQ的內(nèi)存時(shí)，MQ突發(fā)故障宕機(jī)導(dǎo)致消息就丟失了。

?針對(duì)這種生產(chǎn)者投遞數(shù)據(jù)丟失的問(wèn)題，RabbitMQ實(shí)際上是提供了一些機(jī)制的。

比如，有一種重量級(jí)的機(jī)制，就是事務(wù)消息機(jī)制。采用類(lèi)事務(wù)的機(jī)制把消息投遞到MQ，可以保證消息不丟失，但是性能極差，經(jīng)過(guò)測(cè)試性能會(huì)呈現(xiàn)幾百倍的下降。

所以說(shuō)現(xiàn)在一般是不會(huì)用這種過(guò)于重量級(jí)的機(jī)制，而是會(huì)用輕量級(jí)的confirm機(jī)制。

但是我們這篇文章還不能直接講解生產(chǎn)者保證消息不丟失的confirm機(jī)制，因?yàn)檫@種confirm機(jī)制實(shí)際上是采用了類(lèi)似消費(fèi)者的ack機(jī)制來(lái)實(shí)現(xiàn)的。

所以，要深入理解confirm機(jī)制，我們得先從這篇文章開(kāi)始，深?入的分析一下消費(fèi)者手動(dòng)ack機(jī)制保證消息不丟失的底層原理。

二、ack機(jī)制回顧

其實(shí)手動(dòng)ack機(jī)制非常的簡(jiǎn)單，必須要消費(fèi)者確保自己處理完畢了一個(gè)消息，才能手動(dòng)發(fā)送ack給MQ，MQ收到ack之后才會(huì)刪除這個(gè)消息。

如果消費(fèi)者還沒(méi)發(fā)送ack，自己就宕機(jī)了，此時(shí)MQ感知到他的宕機(jī)，就會(huì)重新投遞這條消息給其他的消費(fèi)者實(shí)例。

通過(guò)這種機(jī)制保證消費(fèi)者實(shí)例宕機(jī)的時(shí)候，數(shù)據(jù)是不會(huì)丟失的。

三、ack機(jī)制實(shí)現(xiàn)原理：delivery tag

如果你寫(xiě)好了一個(gè)消費(fèi)者服務(wù)的代碼，讓他開(kāi)始從RabbitMQ消費(fèi)數(shù)據(jù)，這時(shí)這個(gè)消費(fèi)者服務(wù)實(shí)例就會(huì)自己注冊(cè)到RabbitMQ。

所以，RabbitMQ其實(shí)是知道有哪些消費(fèi)者服務(wù)實(shí)例存在的。

大家看看下面的圖，直觀(guān)的感受一下：

接著，RabbitMQ就會(huì)通過(guò)自己內(nèi)部的一個(gè)“basic.delivery”方法來(lái)投遞消息到倉(cāng)儲(chǔ)服務(wù)里去，讓他消費(fèi)消息。

投遞的時(shí)候，會(huì)給這次消息的投遞帶上一個(gè)重要的東西，就是“delivery tag”，你可以認(rèn)為是本次消息投遞的一個(gè)唯一標(biāo)識(shí)。

這個(gè)所謂的唯一標(biāo)識(shí)，有點(diǎn)類(lèi)似于一個(gè)ID，比如說(shuō)消息本次投遞到一個(gè)倉(cāng)儲(chǔ)服務(wù)實(shí)例的唯一ID。通過(guò)這個(gè)唯一ID，我們就可以定位一次消息投遞。

所以這個(gè)delivery tag機(jī)制不要看很簡(jiǎn)單，實(shí)際上他是后面要說(shuō)的很多機(jī)制的核心基礎(chǔ)。

而且這里要給大家強(qiáng)調(diào)另外一個(gè)概念，就是每個(gè)消費(fèi)者從RabbitMQ獲取消息的時(shí)候，都是通過(guò)一個(gè)channel的概念來(lái)進(jìn)行的。

大家回看一下下面的消費(fèi)者代碼片段，我們必須是先對(duì)指定機(jī)器上部署的RabbitMQ建立連接，然后通過(guò)這個(gè)連接獲取一個(gè)channel。

而且如果大家還有點(diǎn)印象的話(huà)，我們?cè)趥}(cāng)儲(chǔ)服務(wù)里對(duì)消息的消費(fèi)、ack等操作，全部都是基于這個(gè)channel來(lái)進(jìn)行的，channel又有點(diǎn)類(lèi)似于是我們跟RabbitMQ進(jìn)行通信的這么一個(gè)句柄，比如看看下面的代碼：

另外這里提一句：之前寫(xiě)那篇文章講解手動(dòng)ack保證數(shù)據(jù)不丟失的時(shí)候，有很多人提出疑問(wèn)：為什么上面代碼里直接是try finally，如果代碼有異常，那還是會(huì)直接執(zhí)行finally里的手動(dòng)ack？其實(shí)很簡(jiǎn)單，自己加上catch就可以了。

好的，咱們繼續(xù)。你大概可以認(rèn)為這個(gè)channel就是進(jìn)行數(shù)據(jù)傳輸?shù)囊粋€(gè)管道吧。對(duì)于每個(gè)channel而言，一個(gè)“delivery tag”就可以唯一的標(biāo)識(shí)一次消息投遞，這個(gè)delivery tag大致而言就是一個(gè)不斷增長(zhǎng)的數(shù)字。

大家來(lái)看看下面的圖，相信會(huì)很好理解的：

如果采用手動(dòng)ack機(jī)制，實(shí)際上倉(cāng)儲(chǔ)服務(wù)每次消費(fèi)了一條消息，處理完畢完成調(diào)度發(fā)貨之后，就會(huì)發(fā)送一個(gè)ack消息給RabbitMQ服務(wù)器，這個(gè)ack消息是會(huì)帶上自己本次消息的delivery tag的。

咱們看看下面的ack代碼，是不是帶上了一個(gè)delivery tag？

channel.basicAck(
delivery.getEnvelope().getDeliveryTag(),
false);

然后，RabbitMQ根據(jù)哪個(gè)channel的哪個(gè)delivery tag，不就可以唯一定位一次消息投遞了？

接下來(lái)就可以對(duì)那條消息刪除，標(biāo)識(shí)為已經(jīng)處理完畢。

這里大家必須注意的一點(diǎn)，就是delivery tag僅僅在一個(gè)channel內(nèi)部是唯一標(biāo)識(shí)消息投遞的。

所以說(shuō)，你ack一條消息的時(shí)候，必須是通過(guò)接受這條消息的同一個(gè)channel來(lái)進(jìn)行。

大家看看下面的圖，直觀(guān)的感受一下。

其實(shí)這里還有一個(gè)很重要的點(diǎn)，就是我們可以設(shè)置一個(gè)參數(shù)，然后就批量的發(fā)送ack消息給RabbitMQ，這樣可以提升整體的性能和吞吐量。

比如下面那行代碼，把第二個(gè)參數(shù)設(shè)置為true就可以了。

channel.basicAck(
delivery.getEnvelope().getDeliveryTag(),
true);

看到這里，大家應(yīng)該對(duì)這個(gè)ack機(jī)制的底層原理有了稍微進(jìn)一步的認(rèn)識(shí)了。起碼是知道delivery tag是啥東西了，他是實(shí)現(xiàn)ack的一個(gè)底層機(jī)制。

然后，我們?cè)賮?lái)簡(jiǎn)單回顧一下自動(dòng)ack、手動(dòng)ack的區(qū)別。

實(shí)際上默認(rèn)用自動(dòng)ack，是非常簡(jiǎn)單的。RabbitMQ只要投遞一個(gè)消息出去給倉(cāng)儲(chǔ)服務(wù)，那么他立馬就把這個(gè)消息給標(biāo)記為刪除，因?yàn)樗遣还軅}(cāng)儲(chǔ)服務(wù)到底接收到?jīng)]有，處理完沒(méi)有的。

所以這種情況下，性能很好，但是數(shù)據(jù)容易丟失。

如果手動(dòng)ack，那么就是必須等倉(cāng)儲(chǔ)服務(wù)完成商品調(diào)度發(fā)貨以后，才會(huì)手動(dòng)發(fā)送ack給RabbitMQ，此時(shí)RabbitMQ才會(huì)認(rèn)為消息處理完畢，然后才會(huì)標(biāo)記消息為刪除。

這樣在發(fā)送ack之前，倉(cāng)儲(chǔ)服務(wù)宕機(jī)，RabbitMQ會(huì)重發(fā)消息給另外一個(gè)倉(cāng)儲(chǔ)服務(wù)實(shí)例，保證數(shù)據(jù)不丟。

四、RabbitMQ如何感知到倉(cāng)儲(chǔ)服務(wù)實(shí)例宕機(jī)

之前就有同學(xué)提出過(guò)這個(gè)問(wèn)題，但是其實(shí)要搞清楚這個(gè)問(wèn)題，其實(shí)不需要深入的探索底層，只要自己大致的思考和推測(cè)一下就可以了。

如果你的倉(cāng)儲(chǔ)服務(wù)實(shí)例接收到了消息，但是沒(méi)有來(lái)得及調(diào)度發(fā)貨，沒(méi)有發(fā)送ack，此時(shí)他宕機(jī)了。

我們想一想就知道，RabbitMQ之前既然收到了倉(cāng)儲(chǔ)服務(wù)實(shí)例的注冊(cè)，因此他們之間必然是建立有某種聯(lián)系的。

一旦某個(gè)倉(cāng)儲(chǔ)服務(wù)實(shí)例宕機(jī)，那么RabbitMQ就必然會(huì)感知到他的宕機(jī)，而且對(duì)發(fā)送給他的還沒(méi)ack的消息，都發(fā)送給其他倉(cāng)儲(chǔ)服務(wù)實(shí)例。

所以這個(gè)問(wèn)題以后有機(jī)會(huì)我們可以深入聊一聊，在這里，大家其實(shí)先建立起來(lái)這種認(rèn)識(shí)即可。

我們?cè)倩仡^看看下面的架構(gòu)圖：

五、倉(cāng)儲(chǔ)服務(wù)處理失敗時(shí)的消息重發(fā)

首先，我們來(lái)看看下面一段代碼：

假如說(shuō)某個(gè)倉(cāng)儲(chǔ)服務(wù)實(shí)例處理某個(gè)消息失敗了，此時(shí)會(huì)進(jìn)入catch代碼塊，那么此時(shí)我們?cè)趺崔k呢？難道還是直接ack消息嗎？

當(dāng)然不是了，你要是還是ack，那會(huì)導(dǎo)致消息被刪除，但是實(shí)際沒(méi)有完成調(diào)度發(fā)貨。

這樣的話(huà)，數(shù)據(jù)不是還是丟失了嗎？因此，合理的方式是使用nack操作。

就是通知RabbitMQ自己沒(méi)處理成功消息，然后讓RabbitMQ將這個(gè)消息再次投遞給其他的倉(cāng)儲(chǔ)服務(wù)實(shí)例嘗試去完成調(diào)度發(fā)貨的任務(wù)。

我們只要在catch代碼塊里加入下面的代碼即可：

channel.basicNack(
delivery.getEnvelope().getDeliveryTag(),
true);

注意上面第二個(gè)參數(shù)是true，意思就是讓RabbitMQ把這條消息重新投遞給其他的倉(cāng)儲(chǔ)服務(wù)實(shí)例，因?yàn)樽约簺](méi)處理成功。

你要是設(shè)置為false的話(huà)，就會(huì)導(dǎo)致RabbitMQ知道你處理失敗，但是還是刪除這條消息，這是不對(duì)的。

同樣，我們還是來(lái)一張圖，大家一起來(lái)感受一下：

六、階段總結(jié)

這篇文章對(duì)之前的ack機(jī)制做了進(jìn)一步的分析，包括底層的delivery tag機(jī)制，以及消息處理失敗時(shí)的消息重發(fā)。

通過(guò)ack機(jī)制、消息重發(fā)等這套機(jī)制的落地實(shí)現(xiàn)，就可以保證一個(gè)消費(fèi)者服務(wù)自身突然宕機(jī)、消息處理失敗等場(chǎng)景下，都不會(huì)丟失數(shù)據(jù)。