一、失敗重試機制

    當消費者出現(xiàn)異常后，消息會不斷requeue(重新入隊）到隊列，再重新發(fā)送給消費者，然后再次異常，再次requeue，無限循環(huán)，導(dǎo)致mq的消息處理飆升，帶來不必要的壓力。lPD28資訊網(wǎng)——每日最新資訊28at.com

    為此，可利用Spring的retry機制，在消費者出現(xiàn)異常時利用本地重試，而不是無限制的requeue到mq隊列。lPD28資訊網(wǎng)——每日最新資訊28at.com

配置參數(shù)如下：lPD28資訊網(wǎng)——每日最新資訊28at.com

listener: # 開啟消費者確認其機制      simple:        prefetch: 1  #消費者每次只能獲取一條消息，處理完才能獲取下一條（可實現(xiàn)能者多勞）        acknowledge-mode: AUTO  # none：關(guān)閉ack；manual：手動ack；auto：自動ack        retry:          enabled: true  #開啟消費者失敗重試          initial-interval: 1000ms  #初始的失敗等待時長為1秒          multiplier: 1 #下次失敗的等待時長倍數(shù)，下次等待時長 = multiplier * last-interval          max-attempts: 3 #最大重試次數(shù)          stateless: true #true無狀態(tài);false有狀態(tài)。如果業(yè)務(wù)中包含事務(wù)，這里改為false

測試結(jié)果：lPD28資訊網(wǎng)——每日最新資訊28at.com

圖片lPD28資訊網(wǎng)——每日最新資訊28at.com

但是重試三次后，隊列里面的消息被踢出了：lPD28資訊網(wǎng)——每日最新資訊28at.com

圖片lPD28資訊網(wǎng)——每日最新資訊28at.com

二、失敗消息處理策略

    在開啟重試模式后，重試次數(shù)耗盡，如果消息依然失敗，則需要有MessageRecoverer接口來處理，它包含三種不同的實現(xiàn):lPD28資訊網(wǎng)——每日最新資訊28at.com

     1、RejectAndDontRequeueRecoverer:重試耗盡后，直接reject，丟棄消息。默認就是這種方式。lPD28資訊網(wǎng)——每日最新資訊28at.com

    2、ImmediateRequeueMessageRecoverer:重試耗盡后，返回nack，消息重新入隊(不建議采用：會出現(xiàn)死循環(huán))。lPD28資訊網(wǎng)——每日最新資訊28at.com

     3、RepublishMessageRecoverer:重試耗盡后，將失敗消息投遞到指定的交換機。(推薦使用)lPD28資訊網(wǎng)——每日最新資訊28at.com

圖片lPD28資訊網(wǎng)——每日最新資訊28at.com

三、實現(xiàn)方式

3.1、定義接收失敗消息的交換機、隊列及其綁定關(guān)系:lPD28資訊網(wǎng)——每日最新資訊28at.com

/**     * 功能描述:定義接收錯誤消費的日志     * @MethodName: receiveErrorMessage     * @MethodParam: [message]     * @Return: void     * @Author: yyalin     * @CreateDate: 2023/11/15 9:55     */    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "errorQueue"),            exchange = @Exchange(name = "errorExchange", type = ExchangeTypes.DIRECT, ignoreDeclarationExceptions = "true"),            key = "errorRouting"    ))    public void receiveErrorMessage(String message) {        log.info("消費者收到發(fā)送錯誤的消息: " + message);    }

3.2、定義RepublishMessageRecoverer:

/** * @Description: TODO:定義錯誤消息接收 * @Author: yyalin * @CreateDate: 2023/11/15 9:58 * @Version: V1.0 */@Configuration@Slf4jpublic class ErrorConfig {    @Bean    public MessageRecoverer messageRecoverer(RabbitTemplate rabbitTemplate){        log.debug("加載RepublishMessageRecoverer");        return new RepublishMessageRecoverer(rabbitTemplate,"errorExchange","errorRouting");    }}

lPD28資訊網(wǎng)——每日最新資訊28at.com

3.3、測試結(jié)果：

圖片lPD28資訊網(wǎng)——每日最新資訊28at.com

3.4、總結(jié)

消費者如何保證消息一定被消費?lPD28資訊網(wǎng)——每日最新資訊28at.com

•     開啟消費者確認機制為auto，由spring確認消息處理成功后返回ack，異常時返回nack。如果一直處理異常會一直重試。
•     開啟消費者失敗重試機制，并設(shè)置MessageRecoverer，多次重試失敗后將消息投遞到異常交換機，交由人工處理。

本文鏈接：http://www.tebozhan.com/showinfo-26-34653-0.htmlRabbitMQ高級之失敗重試機制(含源碼)

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇： 雙向綁定與單向數(shù)據(jù)流之爭，Solid會取代React嗎

下一篇： 新一代WebFlux框架核心技術(shù)Reactor響應(yīng)式編程基本用法