1.?問題引發(fā)

　　由某個服務(wù)BI-collector-xx隊列出現(xiàn)阻塞，影響很整個rabbitMQ集群服務(wù)不可用，多個應(yīng)用MQ生產(chǎn)者服務(wù)出現(xiàn)假死狀態(tài)，系統(tǒng)影響面較廣，業(yè)務(wù)影響很大。當(dāng)時為了應(yīng)急處理，恢復(fù)系統(tǒng)可用，運維相對粗暴的把一堆阻塞隊列信息清空，然后重啟整個集群。

在復(fù)盤整個故障過程中，我心中有不少疑惑，至少存在以下幾個問題點：

1. 為什么出現(xiàn)隊列阻塞？
2. 某個隊列出現(xiàn)阻塞為什么會影響到其他隊列的運行（即多隊列間相互影響）？
3. 某個應(yīng)用MQ隊列出現(xiàn)問題，為什么會導(dǎo)致應(yīng)用不可用呢？

2.?試驗隊列阻塞

某天周末在家里，找個測試環(huán)境，安裝rabbitmq嘗試重現(xiàn)這過程，并做模擬測試。

寫兩個測試應(yīng)用Demo（假設(shè)是兩個項目應(yīng)用）分別有生產(chǎn)者和消費者，并分別使用隊列testA和testB。

為了盡可能還原生產(chǎn)的情況，一開始測試使用了同一個vhost，后面分別設(shè)置不同vhost。

生產(chǎn)者A，示例代碼如下

消費者A

MQ配置

生產(chǎn)者B，每次生產(chǎn)10萬條消息

?

消費者B，代碼故意寫錯（模擬出現(xiàn)異常的情況），不是正常的json串導(dǎo)致解釋json時拋出異常

先了解一下Rabbitmq客戶端啟動連接工作過程，通過wireshark抓包分析，如下

?

先對AMQP做一個簡單的介紹，請求的AMQP協(xié)議方法信息，AMQP協(xié)議方法包含類名+方法名+參數(shù)，這一列主要展示了類名和方法名

• Connection.Start：請求服務(wù)端開始建立連接
• Channel.Open：請求服務(wù)端建立信道
• Queue.Declare：聲明隊列
• Basic.Consume：開始一個消費者，請求指定隊列的消息

詳細(xì)方法可以查看amqp官網(wǎng)https://www.rabbitmq.com/amqp-0-9-1-reference.html

工作過程分析：

Basic.Publish：?客戶端發(fā)送Basic.Publish方法請求，將消息發(fā)布到exchange，rabbitmq server會根據(jù)路由規(guī)則轉(zhuǎn)發(fā)到隊列中；

Basic.Deliver：?服務(wù)端發(fā)送Basic.Deliver方法請求，投遞消息到監(jiān)聽隊列的客戶端消費者；

Basic.Ack：?客戶端發(fā)送Basic.Ack方法請求，告知rabbimq server,消息已接收處理。

兩個應(yīng)用程序啟動后，通過rabbitmq管理控制臺可以觀察一些參數(shù)和監(jiān)控指標(biāo)

?

?

一開始A應(yīng)用生產(chǎn)和消費都是正常的。

B消費端錯誤代碼異常，狂刷報錯信息

?

經(jīng)過大概30分鐘運行，觀察A生產(chǎn)者應(yīng)用控制臺也有出現(xiàn)異常信息

?

查看服務(wù)端連接狀態(tài)出現(xiàn)blocked情況，與生產(chǎn)故障發(fā)生情景很類似。

?

此時客戶端即本機器，CPU和內(nèi)存上漲明顯，風(fēng)扇聲音很響，明顯卡頓，再過30分鐘應(yīng)用基本不可用狀態(tài)。

分析原因

上面錯誤代碼展示了消費者B無法ack，由于沒有進(jìn)行ack導(dǎo)致隊里阻塞。那么問題來了，這是為什么呢？其實這是RabbitMQ的一種保護(hù)機制。防止當(dāng)消息激增的時候，海量的消息進(jìn)入consumer而引發(fā)consumer宕機。

?RabbitMQ提供了一種QOS(服務(wù)質(zhì)量保證)功能，即在非自動確認(rèn)的消息的前提下，限制信道上的消費者所能保持的最大未確認(rèn)的數(shù)量?？梢酝ㄟ^設(shè)置prefetchCount實現(xiàn)，自動確認(rèn)prefetchCount設(shè)置無效。

舉例說明:可以理解為在consumer前面加了一個緩沖容器，容器能容納最大的消息數(shù)量就是PrefetchCount。如果容器沒有滿RabbitMQ就會將消息投遞到容器內(nèi)，如果滿了就不投遞了。當(dāng)consumer對消息進(jìn)行ack以后就會將此消息移除，從而放入新的消息。

通過上面的配置發(fā)現(xiàn)prefetch初始我只配置了2，并且concurrency配置的只有1，所以當(dāng)我發(fā)送了2條錯誤消息以后，由于解析失敗這2條消息一直沒有被ack。將緩沖區(qū)沾滿了，這個時候RabbitMQ認(rèn)為這個consumer已經(jīng)沒有消費能力了就不繼續(xù)給它推送消息了，所以就造成了隊列阻塞。

判斷隊列是否有阻塞的風(fēng)險。

??當(dāng)ack模式為manual，并且線上出現(xiàn)了unacked消息，這個時候不用慌。由于QOS是限制信道channel上的消費者所能保持的最大未確認(rèn)的數(shù)量。所以允許出現(xiàn)unacked的數(shù)量可以通過channelCount * prefetchCount *消費節(jié)點數(shù)量得出。

channlCount就是由concurrency,max-concurrency決定的。

• min = concurrency * prefetch *消費節(jié)點數(shù)量
• max = max-concurrency * prefetch *消費節(jié)點數(shù)量

由此可以得出結(jié)論

• unacked_msg_count < min?隊列不會阻塞。但需要及時處理unacked的消息。
• unacked_msg_count >= min?可能會出現(xiàn)堵塞。
• unacked_msg_count >= max?隊列一定阻塞。

重點注意

1、unacked的消息在consumer切斷連接后(如重啟)再連接，會自動回到隊頭。

2、若將ack模式改成auto自動，這樣會使QOS不生效。會出現(xiàn)大量消息涌入consumer從而可能造成consumer宕機風(fēng)險。

再回看程序配置，做一些分析和調(diào)整

對B消費端問題代碼加個try-catch-finally，不管中間有何問題，都進(jìn)行消息簽收ACK。

?

代碼調(diào)整之后，兩個隊列正常運行，客戶端兩個應(yīng)用也正常運行。

?

?

經(jīng)過一段時間消費，B消費者端已經(jīng)把堆積的消息消費完了。

?

3、????第三個問題原因分析

還是查看抓包信息

Basic.Reject： 客戶端發(fā)送Basic.Reject方法請求，表示無法處理消息，拒絕消息，此時的requeue參數(shù)為true，將消息返回原來的隊列；

Basic.Deliver： 服務(wù)端調(diào)用Basic.Deliver方法，和第一次Basic.Deliver方法不同的是，此時的redeliver參數(shù)為true，表示重新投遞消息到監(jiān)聽隊列的消費者，然后這兩步會一直重復(fù)下去。

RabbitMQ消息監(jiān)聽程序異常時，consumer會向rabbitmq server發(fā)送Basic.Reject，表示消息拒絕接受，由于Spring默認(rèn)requeue-rejected配置為true，消息會重新入隊，然后rabbitmq server重新投遞。就相當(dāng)于死循環(huán)了，所以容易導(dǎo)致消費端資源占用過高，特別是TCP連接數(shù)、線程數(shù)、IO飆升，如果個別程序帶事務(wù)或數(shù)據(jù)庫操作等連接資源得不到釋放也會占滿，導(dǎo)致應(yīng)用假死狀態(tài)（出現(xiàn)問題的時候，查看問題應(yīng)用出現(xiàn)大量的connection timeout錯誤報錯日志）。

因此針對性的，有些業(yè)務(wù)場景（不強調(diào)數(shù)據(jù)強一致性的場景，比如日志收集）可以設(shè)置default-requeue-rejected: false即可。

factory.setDefaultRequeueRejected(false);

??會根據(jù)異常類型選擇直接丟棄或加入dead-letter-exchange中。

消費者端正確的使用手動確認(rèn)示例結(jié)構(gòu)代碼，很重要！

try {// 業(yè)務(wù)邏輯。
}catch (Exception e){// 輸出錯誤日志。
}finally {// 消息簽收。
}

4、????驗證隊列設(shè)置最大長度限制

設(shè)置queueLengthLimit隊列最大長度限制 x-max-length=5

?

生產(chǎn)者原本想要生產(chǎn)10條消息

?

由于受到隊列最大長度限制，實際上只有5條入隊列里面。

?

消費者拿出來的消息，僅有5條，從NO.6~NO.10

改變消費者程序，讓生產(chǎn)者一直產(chǎn)生消息，消費者消費速度明顯趕不上生產(chǎn)者的生產(chǎn)速度。

?

?

從消費端來看消息是隨機性入隊的，隊列里面一直最多5條消息，發(fā)再多也進(jìn)不了，消息者和生產(chǎn)者也不會發(fā)生什么異常，只是消息會隨機性丟失（并沒有全部入隊）。

運行情況良好，除了消息沒有全部入隊列 ，沒有出現(xiàn)異常情況

?

消費比較慢，本機器CPU和內(nèi)存各項指標(biāo)正常，沒有異常。

搞一個異常情況出現(xiàn)unack，最大隊列長度限制，是不算unack數(shù)量的，如下圖所示

?

異常之后，此觀察MQ監(jiān)控管理后臺

?

生產(chǎn)者不停一直在生產(chǎn)消息，運行30分鐘，觀察生產(chǎn)者應(yīng)用也是正常的的，就是消息入不了隊列。

?

?

5、??檢查實際的業(yè)務(wù)端代碼

再看我們業(yè)務(wù)系統(tǒng)消費端代碼，消費端各種不規(guī)范寫法都有，以下例舉幾個典型

1、手動簽收有ACK，但是沒有try-catch-finally結(jié)構(gòu)，消費端業(yè)務(wù)代碼如下：

2、有try-catch-finally結(jié)構(gòu)，但是deliverTag是一個固定值0，一樣的會出問題。

?

3、自動簽收確認(rèn)的，大量消息的時候，容易搞死消費端應(yīng)用。

?

6.?總結(jié)

• 生產(chǎn)環(huán)境不建議使用自動ack模式，這樣會使QOS無法生效。
• 在使用手動ack的時候，需要非常注意消息簽收，業(yè)務(wù)代碼使用try-catch-finally處理結(jié)構(gòu)，防止業(yè)務(wù)代碼異常時無法簽收。
• 規(guī)范約束mq客戶端代碼，正確的使用Rabbitmq配置。
• 不同業(yè)務(wù)項目設(shè)置不同的vhost可以隔離一些影響，提升rabbitmq資源使用。
• 考慮設(shè)置dead-letter-exchange，當(dāng)設(shè)置了?requeue=false時，可以放入dead-letter-exchange，可以快速排查定位問題。
• Exchange和隊列的最大長度限制可以是限制消息的數(shù)量（參數(shù)：x-max-length），或者是消息的總字節(jié)數(shù)（總字節(jié)數(shù)表示的是所有的消息體的字節(jié)數(shù)，忽略消息的屬性和任何頭部信息），又或者兩者都進(jìn)行了限制，兩者取小值生效，只有處于ready狀態(tài)的消息被計數(shù)，未被確認(rèn)的消息不會被計數(shù)受到limit的限制。最大隊列設(shè)置可以限制生產(chǎn)端，但會造成消息丟失風(fēng)險，最大消息數(shù)量限制，不能完全解決隊列阻塞問題。
• 盡量使用Direct-exchange，Direct 類型的 Exchange 投遞消息是最快的。
  • Direct：處理路由鍵，需要將一個隊列綁定到交換機上，要求該消息與一個特定的路由鍵完全匹配。這是一個完整的匹配。如果一個隊列綁定到該交換機上要求路由鍵為“A”，則只有路由鍵為“A”的消息才被轉(zhuǎn)發(fā)，不會轉(zhuǎn)發(fā)路由鍵為"B"，只會轉(zhuǎn)發(fā)路由鍵為“A”；
  • Topic：將路由鍵和某模式進(jìn)行匹配。此時隊列需要綁定要一個模式上。符號“#”匹配一個或多個詞，符號“*”只能匹配一個詞；
  • Fanout：不處理路由鍵。只需要簡單的將隊列綁定到交換機上。一個發(fā)送到該類型交換機的消息都會被廣播到與該交換機綁定的所有隊列上；
  • Headers：不處理路由鍵，而是根據(jù)發(fā)送的消息內(nèi)容中的 headers 屬性進(jìn)行匹配。在綁定 Queue 與 Exchange 時指定一組鍵值對；當(dāng)消息發(fā)送到 RabbitMQ 時會取到該消息的 headers 與 Exchange 綁定時指定的鍵值對進(jìn)行匹配；如果完全匹配則消息會路由到該隊列，否則不會路由到該隊列。