生產(chǎn)RabbitMQ隊(duì)列阻塞該如何處理？

image

前言

??那天我和同事一起吃完晚飯回公司加班，然后就群里就有人@我說(shuō)xxx商戶說(shuō)收不到推送，一開(kāi)始覺(jué)得沒(méi)啥。我第一反應(yīng)是不是極光沒(méi)注冊(cè)上，就讓客服通知商戶，重新登錄下試試。這邊打開(kāi)極光推送的后臺(tái)進(jìn)行檢查。后面反應(yīng)收不到推送的越來(lái)越多，我就知道這事情不簡(jiǎn)單。

事故經(jīng)過(guò)

??由于大量商戶反應(yīng)收不到推送，我第一反應(yīng)是不是推送系統(tǒng)掛了，導(dǎo)致沒(méi)有進(jìn)行推送。于是讓運(yùn)維老哥檢查推送系統(tǒng)各節(jié)點(diǎn)的情況，發(fā)現(xiàn)都正常。于是打開(kāi)RabbitMQ的管控臺(tái)看了一下，人都蒙了。已經(jīng)有幾萬(wàn)條消息處于ready狀態(tài)，還有幾百條unacked的消息。

??我以為推送服務(wù)和MQ連接斷開(kāi)了，導(dǎo)致無(wú)法推送消息，于是讓運(yùn)維重啟推送服務(wù)，將所有的推送服務(wù)重啟完，發(fā)現(xiàn)unacked的消息全部變成ready，但是沒(méi)過(guò)多久又有幾百條unacked的消息了，這個(gè)就很明顯了能消費(fèi)，沒(méi)有進(jìn)行ack呀。

??當(dāng)時(shí)我以為是網(wǎng)絡(luò)問(wèn)題，導(dǎo)致mq無(wú)法接收到ack，讓運(yùn)維老哥檢查了一下，發(fā)現(xiàn)網(wǎng)絡(luò)沒(méi)問(wèn)題?，F(xiàn)在看是真的是傻，網(wǎng)絡(luò)有問(wèn)題連接都連不上。由于確定的是無(wú)法ack造成的，立馬將ack模式由原來(lái)的manual 改成auto緊急發(fā)布。將所有的節(jié)點(diǎn)升級(jí)好以后，發(fā)現(xiàn)推送正常了。

??你以為這就結(jié)束了其實(shí)并沒(méi)有，沒(méi)過(guò)多久發(fā)現(xiàn)有一臺(tái)MQ服務(wù)出現(xiàn)異常，由于生產(chǎn)采用了鏡像隊(duì)列，立即將這臺(tái)有問(wèn)題的MQ從集群中移除。直接進(jìn)行重置，然后加入回集群。這事情算是告一段落了。此時(shí)已經(jīng)接近24:00了。

image

??時(shí)間來(lái)到第二天上午10:00，運(yùn)維那邊又出現(xiàn)報(bào)警了，說(shuō)推送系統(tǒng)有臺(tái)機(jī)器，磁盤(pán)快被寫(xiě)滿了，并且占用率很高。我的乖乖從昨晚到現(xiàn)在寫(xiě)了快40G的日志，一看報(bào)錯(cuò)信息瞬間就明白問(wèn)題出在哪里了。麻溜的把bug修了緊急發(fā)布。

吐槽一波公司的ELK，壓根就沒(méi)有收集到這個(gè)報(bào)錯(cuò)信息，導(dǎo)致我沒(méi)有及時(shí)發(fā)現(xiàn)。

image

事故重現(xiàn)-隊(duì)列阻塞

MQ配置

spring:
  # 消息隊(duì)列
  rabbitmq:
    host: 10.0.0.53
    username: guest
    password: guest
    virtual-host: local
    port: 5672
    # 消息發(fā)送確認(rèn)
    publisher-confirm-type: correlated
    # 開(kāi)啟發(fā)送失敗退回
    publisher-returns: true
    listener:
      simple:
        # 消費(fèi)端最小并發(fā)數(shù)
        concurrency: 1
        # 消費(fèi)端最大并發(fā)數(shù)
        max-concurrency: 5
        # 一次請(qǐng)求中預(yù)處理的消息數(shù)量
        prefetch: 2
        # 手動(dòng)應(yīng)答
        acknowledge-mode: manual

問(wèn)題代碼

@RabbitListener(queues = ORDER_QUEUE)
public void receiveOrder(@Payload String encryptOrderDto,
                                      @Headers Map<String,Object> headers,
                                      Channel channel) throws Exception {
    // 解密和解析
    String decryptOrderDto = EncryptUtil.decryptByAes(encryptOrderDto);
    OrderDto orderDto = JSON.parseObject(decryptOrderDto, OrderDto.class);

    try {
        // 模擬推送
        pushMsg(orderDto);
    }catch (Exception e){
        log.error("推送失敗-錯(cuò)誤信息:{},消息內(nèi)容:{}", e.getLocalizedMessage(), JSON.toJSONString(orderDto));
    }finally {
        // 消息簽收
        channel.basicAck((Long) headers.get(AmqpHeaders.DELIVERY_TAG),false);
    }

}

??看起來(lái)好像沒(méi)啥問(wèn)題。由于和交易系統(tǒng)約定好，訂單數(shù)據(jù)需要先轉(zhuǎn)換json串，然后再使用AES進(jìn)行加密，所以這邊需要，先進(jìn)行解密然后在進(jìn)行解析。才能得到訂單數(shù)據(jù)。

??為了防止消息丟失，交易系統(tǒng)做了失敗重發(fā)機(jī)制，防止消息丟失，不巧的是重發(fā)的時(shí)候沒(méi)有對(duì)訂單數(shù)據(jù)進(jìn)行加密。這就導(dǎo)致推送系統(tǒng)，在解密的時(shí)候出異常，從而無(wú)法進(jìn)行ack。

默默的吐槽一句：人在家中坐，鍋從天上來(lái)。

模擬推送

推送代碼

發(fā)送3條正常的消息

curl http://localhost:8080/sendMsg/3

發(fā)送1條錯(cuò)誤的消息

curl http://localhost:8080/sendErrorMsg/1

再發(fā)送3條正常的消息

curl http://localhost:8080/sendMsg/3
image

??觀察日志發(fā)下，雖然有報(bào)錯(cuò)，但是還能正常進(jìn)行推送。但是RabbitMQ已經(jīng)出現(xiàn)了一條unacked的消息。

image

繼續(xù)發(fā)送1條錯(cuò)誤的消息

curl http://localhost:8080/sendErrorMsg/1

再發(fā)送3條正常的消息

curl http://localhost:8080/sendMsg/3

??這個(gè)時(shí)候你會(huì)發(fā)現(xiàn)控制臺(tái)報(bào)錯(cuò)，當(dāng)然錯(cuò)誤信息是解密失敗，但是正常的消息卻沒(méi)有被消費(fèi)，這個(gè)時(shí)候其實(shí)隊(duì)列已經(jīng)阻塞了。

image
image

??從RabbitMQ管控臺(tái)也可以看到，剛剛發(fā)送的的3條消息處于ready狀態(tài)。這個(gè)時(shí)候就如果一直有消息進(jìn)入，都會(huì)堆積在隊(duì)里里面無(wú)法被消費(fèi)。

再發(fā)送3條正常的消息

curl http://localhost:8080/sendMsg/3
image

分析原因

??上面說(shuō)了是由于沒(méi)有進(jìn)行ack導(dǎo)致隊(duì)里阻塞。那么問(wèn)題來(lái)了，這是為什么呢？其實(shí)這是RabbitMQ的一種保護(hù)機(jī)制。防止當(dāng)消息激增的時(shí)候，海量的消息進(jìn)入consumer而引發(fā)consumer宕機(jī)。

??RabbitMQ提供了一種QOS(服務(wù)質(zhì)量保證)功能，即在非自動(dòng)確認(rèn)的消息的前提下，限制信道上的消費(fèi)者所能保持的最大未確認(rèn)的數(shù)量。可以通過(guò)設(shè)置PrefetchCount實(shí)現(xiàn)。

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

listener:
  simple:
    # 消費(fèi)端最小并發(fā)數(shù)
    concurrency: 1
    # 消費(fèi)端最大并發(fā)數(shù)
    max-concurrency: 5
    # 一次處理的消息數(shù)量
    prefetch: 2
    # 手動(dòng)應(yīng)答
    acknowledge-mode: manual

prefetch參數(shù)就是PrefetchCount

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

判斷隊(duì)列是否有阻塞的風(fēng)險(xiǎn)。

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

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

  • min = concurrency * prefetch * 節(jié)點(diǎn)數(shù)量
  • max = max-concurrency * prefetch * 節(jié)點(diǎn)數(shù)量

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

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

這里需要好好理解一下。

處理方法

其實(shí)處理的方法很簡(jiǎn)單，將解密和解析的方法放入try catch中就解決了這樣不管解密正常與否，消息都會(huì)被簽收。如果出錯(cuò)將會(huì)輸出錯(cuò)誤日志，讓開(kāi)發(fā)人員進(jìn)行處理了。

對(duì)于這個(gè)就需要有日志監(jiān)控系統(tǒng)，來(lái)及時(shí)告警了。

@RabbitListener(queues = ORDER_QUEUE)
public void receiveOrder(@Payload String encryptOrderDto,
                                      @Headers Map<String,Object> headers,
                                      Channel channel) throws Exception {
    try {

        // 解密和解析
        String decryptOrderDto = EncryptUtil.decryptByAes(encryptOrderDto);
        OrderDto orderDto = JSON.parseObject(decryptOrderDto, OrderDto.class);
        
        // 模擬推送
        pushMsg(orderDto);
    }catch (Exception e){
        log.error("推送失敗-錯(cuò)誤信息:{},消息內(nèi)容:{}", e.getLocalizedMessage(), encryptOrderDto);
    }finally {
        // 消息簽收
        channel.basicAck((Long) headers.get(AmqpHeaders.DELIVERY_TAG),false);
    }

}

注意的點(diǎn)

??unacked的消息在consumer切斷連接后(重啟)，會(huì)自動(dòng)回到隊(duì)頭。

事故重現(xiàn)-磁盤(pán)占用飆升

??一開(kāi)始我不知道代碼有問(wèn)題，就是以為單純的沒(méi)有進(jìn)行ack所以將ack模式改成auto自動(dòng)，緊急升級(jí)了，這樣不管正常與否，消息都會(huì)被簽收，所以在當(dāng)時(shí)確實(shí)是解決了問(wèn)題。

??其實(shí)現(xiàn)在回想起來(lái)是非常危險(xiǎn)的操作的，將ack模式改成auto自動(dòng)，這樣會(huì)使QOS不生效。會(huì)出現(xiàn)大量消息涌入consumer從而造成consumer宕機(jī)，可以是因?yàn)楫?dāng)時(shí)在晚上，交易比較少，并且推送系統(tǒng)有多個(gè)節(jié)點(diǎn)，才沒(méi)出現(xiàn)問(wèn)題。

問(wèn)題代碼

@RabbitListener(queues = ORDER_QUEUE)
public void receiveOrder(@Payload String encryptOrderDto,
                                      @Headers Map<String,Object> headers,
                                      Channel channel) throws Exception {
    // 解密和解析
    String decryptOrderDto = EncryptUtil.decryptByAes(encryptOrderDto);
    OrderDto orderDto = JSON.parseObject(decryptOrderDto, OrderDto.class);
    
    try {

        // 模擬推送
        pushMsg(orderDto);
    }catch (Exception e){
        log.error("推送失敗-錯(cuò)誤信息:{},消息內(nèi)容:{}", e.getLocalizedMessage(), encryptOrderDto);
    }finally {
        // 消息簽收
        channel.basicAck((Long) headers.get(AmqpHeaders.DELIVERY_TAG),false);
    }

}

配置文件

listener:
  simple:
    # 消費(fèi)端最小并發(fā)數(shù)
    concurrency: 1
    # 消費(fèi)端最大并發(fā)數(shù)
    max-concurrency: 5
    # 一次處理的消息數(shù)量
    prefetch: 2
    # 手動(dòng)應(yīng)答
    acknowledge-mode: auto

??由于當(dāng)時(shí)不知道交易系統(tǒng)的重發(fā)機(jī)制，重發(fā)時(shí)沒(méi)有對(duì)訂單數(shù)據(jù)加密的bug，所以還是會(huì)發(fā)出少量有誤的消息。

發(fā)送1條錯(cuò)誤的消息

curl http://localhost:8080/sendErrorMsg/1
image
image

原因

??RabbitMQ消息監(jiān)聽(tīng)程序異常時(shí)，consumer會(huì)向rabbitmq server發(fā)送Basic.Reject，表示消息拒絕接受，由于Spring默認(rèn)requeue-rejected配置為true，消息會(huì)重新入隊(duì)，然后rabbitmq server重新投遞。就相當(dāng)于死循環(huán)了，所以控制臺(tái)在瘋狂刷錯(cuò)誤日志造成磁盤(pán)利用率飆升的原因。

解決方法

??將default-requeue-rejected: false即可。

總結(jié)

  • 個(gè)人建議，生產(chǎn)環(huán)境不建議使用自動(dòng)ack，這樣會(huì)QOS無(wú)法生效。
  • 在使用手動(dòng)ack的時(shí)候，需要非常注意消息簽收。
  • 其實(shí)在將有問(wèn)題的MQ重置時(shí)，是將錯(cuò)誤的消息給清除才沒(méi)有問(wèn)題了，相當(dāng)于是消息丟失了。
try {
    // 業(yè)務(wù)邏輯。
}catch (Exception e){
    // 輸出錯(cuò)誤日志。
}finally {
    // 消息簽收。
}

參考資料

代碼地址

https://gitee.com/huangxunhui/rabbitmq_accdient.git

結(jié)尾

??如果有人告訴你遇到線上事故不要慌，除非是超級(jí)大佬久經(jīng)沙場(chǎng)。否則就是瞎扯淡，你讓他來(lái)試試，看看他會(huì)不會(huì)大腦一片空白，直冒汗。

??如果覺(jué)得對(duì)你有幫助，可以多多評(píng)論，多多點(diǎn)贊哦，也可以到我的主頁(yè)看看，說(shuō)不定有你喜歡的文章，也可以隨手點(diǎn)個(gè)關(guān)注哦，謝謝。

??我是不一樣的科技宅，每天進(jìn)步一點(diǎn)點(diǎn)，體驗(yàn)不一樣的生活。我們下期見(jiàn)！

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

友情鏈接更多精彩內(nèi)容