一、持久化概念

在上一章內(nèi)容中我們已經(jīng)看到了如何處理任務(wù)不丟失的情況，但是如何保障當(dāng) RabbitMQ 服務(wù)停掉后消 息生產(chǎn)者發(fā)送過來(lái)的消息不丟失呢？默認(rèn)情況下 RabbitMQ 退出或由于某種原因崩潰時(shí)，它將忽視隊(duì)列 和消息，除非告知它不要這樣做。

確保消息不會(huì)丟失需要做兩件事：我們需要將隊(duì)列和消息都標(biāo) 記為持久化。

二、實(shí)現(xiàn)持久化

2.1 隊(duì)列實(shí)現(xiàn)持久化

如果要隊(duì)列實(shí)現(xiàn)持久化，需要在聲明隊(duì)列的時(shí)候把 durable 參數(shù)設(shè)置為持久化 。

boolean durable = true;
channel.queueDeclare(ACK_QUEUE_NAME, durable, false, false, null);

但是需要注意的就是如果之前聲明的隊(duì)列不是持久化的，需要把原先隊(duì)列先刪除，或者重新 創(chuàng)建一個(gè)持久化的隊(duì)列，不然就會(huì)出現(xiàn)錯(cuò)誤 。

當(dāng)我們將durable參數(shù)位置傳遞true之后，即使重啟 rabbitmq 隊(duì)列也依然存在。

2.2 消息實(shí)現(xiàn)持久化

要想讓消息實(shí)現(xiàn)持久化需要在消息生產(chǎn)者修改代碼，需要像channel的basicPublish方法中props屬性中傳遞MessageProperties.PERSISTENT_TEXT_PLAIN 參數(shù)。

將消息標(biāo)記為持久化并不能保證完全不丟失消息。盡管它告訴 RabbitMQ 將消息保存到磁盤，但是這里依然存在當(dāng)消息剛準(zhǔn)備存儲(chǔ)在磁盤，但是還沒有存儲(chǔ)完，事實(shí)上消息還在緩存的一個(gè)時(shí)間點(diǎn)。在這個(gè)時(shí)候消息并沒有真正寫入磁盤，持久性保證并不強(qiáng)。但是對(duì)于我們的簡(jiǎn)單任務(wù)隊(duì)列而言，這已經(jīng)綽綽有余了。后面的發(fā)布確認(rèn)章節(jié)將介紹更強(qiáng)有力的持久化策略。

沒有持久化：

channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));

實(shí)現(xiàn)持久化:

channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

三、不公平分發(fā)

在最開始的時(shí)候我們學(xué)習(xí)到 RabbitMQ 分發(fā)消息采用的輪訓(xùn)分發(fā)，但是在某種場(chǎng)景下這種策略并不是 很好，比方說(shuō)有兩個(gè)消費(fèi)者在處理任務(wù)，其中有個(gè)消費(fèi)者 1 處理任務(wù)的速度非?？?，而另外一個(gè)消費(fèi)者 2 處理速度卻很慢，這個(gè)時(shí)候我們還是采用輪訓(xùn)分發(fā)的話就會(huì)到這處理速度快的這個(gè)消費(fèi)者很大一部分時(shí)間 處于空閑狀態(tài)，而處理慢的那個(gè)消費(fèi)者一直在干活，這種分配方式在這種情況下就不合理，但是 RabbitMQ 并不知道這種情況，它依然很公平地進(jìn)行分發(fā)。

為了避免這種情況，我們可以設(shè)置參數(shù) channel.basicQos(1);

意思就是如果這個(gè)任務(wù)我還沒有處理完或者我還沒有應(yīng)答你，你先別分配給我，我目前只能處理一個(gè) 任務(wù)，然后 rabbitmq 就會(huì)把該任務(wù)分配給沒有那么忙的那個(gè)空閑消費(fèi)者，當(dāng)然如果所有的消費(fèi)者都沒有完 成手上任務(wù)，隊(duì)列還在不停的添加新任務(wù)，隊(duì)列有可能就會(huì)遇到隊(duì)列被撐滿的情況，這個(gè)時(shí)候就只能添加 新的 worker 或者改變其他存儲(chǔ)任務(wù)的策略。

四、預(yù)取值

本身消息的發(fā)送就是異步發(fā)送的，所以在任何時(shí)候，channel 上肯定不止只有一個(gè)消息，另外來(lái)自消費(fèi) 者的手動(dòng)確認(rèn)本質(zhì)上也是異步的。因此這里就存在一個(gè)未確認(rèn)的消息緩沖區(qū)，因此希望開發(fā)人員能限制此 緩沖區(qū)的大小，以避免緩沖區(qū)里面無(wú)限制的未確認(rèn)消息問題。

這個(gè)時(shí)候就可以通過使用 basic.qos 方法設(shè) 置“預(yù)取計(jì)數(shù)”值來(lái)完成。該值定義通道上允許的未確認(rèn)消息的最大數(shù)量。一旦數(shù)量達(dá)到配置的數(shù)量， RabbitMQ 將停止在通道上傳遞更多消息，除非至少有一個(gè)未處理的消息被確認(rèn)。

設(shè)置預(yù)取值的方式和三中一樣，即channel.basicQos(預(yù)取值)。

例如，假設(shè)在通道上有 未確認(rèn)的消息 5、6、7，8，并且通道的預(yù)取計(jì)數(shù)設(shè)置為 4，此時(shí) RabbitMQ 將不會(huì)在該通道上再傳遞任何 消息，除非至少有一個(gè)未應(yīng)答的消息被 ack。比方說(shuō) tag=6 這個(gè)消息剛剛被確認(rèn) ACK，RabbitMQ 將會(huì)感知 這個(gè)情況并再發(fā)送一條消息。

消息應(yīng)答和 QoS 預(yù)取值對(duì)用戶吞吐量有重大影響。通常，增加預(yù)取值將提高 向消費(fèi)者傳遞消息的速度。雖然自動(dòng)應(yīng)答傳輸消息速率是最佳的，但是，在這種情況下已傳遞但尚未處理 的消息的數(shù)量也會(huì)增加，從而增加了消費(fèi)者的 RAM 消耗(隨機(jī)存取存儲(chǔ)器)。應(yīng)該小心使用具有無(wú)限預(yù)處理 的自動(dòng)確認(rèn)模式或手動(dòng)確認(rèn)模式，消費(fèi)者消費(fèi)了大量的消息如果沒有確認(rèn)的話，會(huì)導(dǎo)致消費(fèi)者連接節(jié)點(diǎn)的 內(nèi)存消耗變大，所以找到合適的預(yù)取值是一個(gè)反復(fù)試驗(yàn)的過程，在不同的負(fù)載下該值取值也不同，100 到 300 范 圍內(nèi)的值通?？商峁┳罴训耐掏铝?，并且不會(huì)給消費(fèi)者帶來(lái)太大的風(fēng)險(xiǎn)。預(yù)取值為 1 是最保守的。當(dāng)然這 將使吞吐量變得很低，特別是消費(fèi)者連接延遲很嚴(yán)重的情況下，特別是在消費(fèi)者連接等待時(shí)間較長(zhǎng)的環(huán)境 中。對(duì)于大多數(shù)應(yīng)用來(lái)說(shuō)，稍微高一點(diǎn)的值將是最佳的。 文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-601582.html