解決RabbitMq消息隊列Qos?Prefetch消息堵塞問題

更新時間：2022年01月31日 10:40:30 作者：james_searcher

這篇文章主要為大家介紹了關(guān)于如何解決解決RabbitMq?Qos?Prefetch消息堵塞的問題分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助

mq是實現(xiàn)代碼擴(kuò)展的有利手段，個人喜歡用概念來學(xué)習(xí)新知識，介紹堵塞問題的之前，先來段概念的學(xué)習(xí)。

ConnectionFactory：創(chuàng)建connection的工廠類

Connection: 簡單理解為socket

Channel:和mq交互的接口，定義queue、exchange和綁定queue、exhange等接口都是它。

接下來就是和mq的交互類

exchange:簡單地看成路由，類型不是重點(diǎn)，看看官網(wǎng)即可

queue:客戶端監(jiān)聽的是queue，而不是exchange，但是使用queue的前提要先將exchange和queue綁定。用過java queue工具類應(yīng)該很容易上手，queue分為寫和讀，各自可以有自己頻率，寫得快讀得慢，容易堵塞；寫得慢讀得快又容易造成消費(fèi)者的空閑。

Prefetc:一個重要卻容易被忽略的指標(biāo)，也是這次遇到的問題。

prefetch與消息投遞

prefetch是指單一消費(fèi)者最多能消費(fèi)的unacked messages數(shù)目。

如何理解呢？

mq為每一個 consumer設(shè)置一個緩沖區(qū)，大小就是prefetch。每次收到一條消息，MQ會把消息推送到緩存區(qū)中，然后再推送給客戶端。當(dāng)收到一個ack消息時(consumer 發(fā)出baseack指令)，mq會從緩沖區(qū)中空出一個位置，然后加入新的消息。但是這時候如果緩沖區(qū)是滿的，MQ將進(jìn)入堵塞狀態(tài)。

更具體點(diǎn)描述，假設(shè)prefetch值設(shè)為10，共有兩個consumer。也就是說每個consumer每次會從queue中預(yù)抓取 10 條消息到本地緩存著等待消費(fèi)。同時該channel的unacked數(shù)變?yōu)?0。而Rabbit投遞的順序是，先為consumer1投遞滿10個message，再往consumer2投遞10個message。如果這時有新message需要投遞，先判斷channel的unacked數(shù)是否等于20，如果是則不會將消息投遞到consumer中，message繼續(xù)呆在queue中。之后其中consumer對一條消息進(jìn)行ack，unacked此時等于19，Rabbit就判斷哪個consumer的unacked少于10，就投遞到哪個consumer中。

我遇到的問題是一個粗心的程序員，在編寫代碼的時候，他對某些消息處理方式是這樣的

  if (success) {
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    } else {
                        logger.error("######### The message is not delete from queue : {}", body);
                    }

首先他講ack機(jī)制設(shè)置為手動的，然后他的理解是如果處理成功的消息，就ack給MQ，期望MQ就可以刪除完成的數(shù)據(jù)。不然，保留數(shù)據(jù)再次被處理。

這里的誤區(qū)就是就是對ack的理解，失敗的時候，如果需要讓程序繼續(xù)處理，應(yīng)該使用basicNack，并告訴mq將消息再次放入隊列

    if (success) {
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                        } else {
                            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                        }

對于客戶端意外宕機(jī)的情況，沒有ack服務(wù)器確實不會刪除掉數(shù)據(jù)，但是consumer重啟以后，對于服務(wù)器就是一個新的消費(fèi)者了，也就是它的緩沖區(qū)又被重置為原來的n-prefetch，所以這個問題被粗心的小哥想當(dāng)然地測試通過了。

prefetch的大小應(yīng)該為多少

這篇文章給了很好的建議，我簡單地說一下我的理解。

理想狀況下，計算MQ SERVER 從緩沖區(qū)中拿到消息并推送到消費(fèi)端，加上消費(fèi)端處理完ack消息到MQ server,的時間，假設(shè)為100ms，其中消費(fèi)端處理業(yè)務(wù)話費(fèi)了10ms。

這里可以得出我們 prefetch = 100ms / 10ms = 10，也就是消息來回的總時間/業(yè)務(wù)處理的時間，這里要求我們 prefetch >= 10。一般計算這個時間不會太準(zhǔn)確只能毛姑姑的，所以prefetch一般要大一點(diǎn)。但是這個值也不能太大，不然消費(fèi)端就一只處于空閑狀態(tài)了。

所以如果你保證所有的消息都ack了，但是還是出現(xiàn)比較長時間的堵塞，你就或者加大一點(diǎn)prefetch，或者多加一些機(jī)器，或者減少業(yè)務(wù)處理的時間了。一開始建議采用或者，使用一個線程池來處理這些業(yè)務(wù)邏輯。

以上就是解決RabbitMqQosPrefetch消息堵塞問題的詳細(xì)內(nèi)容，更多關(guān)于RabbitMqQosPrefetch消息堵塞的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: