2. 程式人生 > >RocketMQ重試機制和訊息冪等

RocketMQ重試機制和訊息冪等

阿新 • • 發佈：2020-02-18

一、重試機制

　　由於MQ經常處於複雜的分散式系統中，考慮網路波動，服務宕機，程式異常因素，很有可能出現訊息傳送或者消費失敗的問題。因此，訊息的重試就是所有MQ中介軟體必須考慮到的一個關鍵點。如果沒有訊息重試，就可能產生訊息丟失的問題，可能對系統產生很大的影響。所以，秉承寧可多發訊息，也不可丟失訊息的原則，大部分MQ都對訊息重試提供了很好的支援。

　　MQ 消費者的消費邏輯失敗時，可以通過設定返回狀態達到訊息重試的結果。

　　MQ 訊息重試只針對叢集消費方式生效；廣播方式不提供失敗重試特性，即消費失敗後，失敗訊息不再重試，繼續消費新的訊息。

1、模擬異常

消費者：

package com.zn.retry;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * RocketMQ重試機制消費者
 */
public class RetryConsumer {
    public static void main(String[] args) throws MQClientException {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設定NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設定例項名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic","TagA");

        //監聽訊息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //獲取訊息
                for (MessageExt messageExt:list){
                    System.out.println(messageExt.getMsgId()+"---"+new String(messageExt.getBody()));
                }
                try {
                    //模擬錯誤
                    int i=5/0;
                }catch (Exception e){
                    e.printStackTrace();
                    //需要重試
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //不需要重試
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //啟動消費者
        consumer.start();
        System.out.println("Consumer Started!");
    }
}

控制檯效果：

　　

2、模擬網路延遲

package com.zn.retry;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * RocketMQ重試機制消費者
 */
public class RetryConsumer {
    public static void main(String[] args) throws MQClientException {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設定NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設定例項名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic","TagA");

        //監聽訊息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //獲取訊息
                for (MessageExt messageExt:list){
                    System.out.println(messageExt.getMsgId()+"---"+new String(messageExt.getBody()));
                }
                try {
                    //網路延遲
                    Thread.sleep(600000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //消費成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //啟動消費者
        consumer.start();
        System.out.println("Consumer Started!");
    }
}

二、訊息冪等

1、在什麼情況下會發生RocketMQ的訊息重複消費

　　 ①、當系統的呼叫鏈路比較長的時候，比如系統A呼叫系統B，系統B再把訊息傳送到RocketMQ中，在系統A呼叫系統B的時候，如果系統B處理成功，但是遲遲沒有將呼叫成功的結果返回給系統A的時候，系統A就會嘗試重新發起請求給系統B，造成系統B重複處理，發起多條訊息給RocketMQ造成重複消費；

　　②、在系統B傳送給RocketMQ的時候，也有可能會發生和上面一樣的問題，訊息傳送超時，節骨系統B重試，導致RocketMQ接收到了重讀訊息；

　　③、當RocketMQ成功接收到訊息，並將訊息交給消費者處理，如果消費者消費完成後還沒來得及提交offset給RocketMQ，自己宕機或者重啟了，那麼RocketMQ沒有接收到offset，就會認為消費失敗了，會重發訊息給消費者再次消費；

2、如何解決訊息的重複消費

　　通過冪等性來保證，只要保證重複訊息不對結果產生影響，就完美地解決這個問題。

在生產者端保證冪等性，一下兩種方式：

　　①、RocketMQ支援訊息查詢的功能，只要去RocketMQ查詢一下是否已經發送過該條訊息就可以了，不存在則傳送，存在則不傳送；

　　②、引入Redis，在傳送訊息到RocketMQ成功之後，向Redis中插入一條資料，如果傳送重試，則先去Redis查詢一個該條訊息是否已經發送過了，存在的話就不重複傳送訊息了；

　　方法一：RocketMQ訊息查詢的效能不是特別好，如果在高併發的場景下，每條訊息在傳送到RocketMQ時都去查詢一下，可能會影響介面的效能；

　　方法二：在一些極端的場景下，Redis也無法保證訊息傳送成功之後，就一定能寫入Redis成功，比如寫入訊息成功而Redis此時宕機，那麼再次查詢Redis判斷訊息是否已經發送過，是無法得到正確結果的；

3、生產者

package com.zn.idempotent;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

/**
 * 訊息冪等生產者
 */
public class IdempotentProvider {
    public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
        //建立一個生產者
        DefaultMQProducer producer=new DefaultMQProducer("rmq-group");
        //設定NameServer地址
        producer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設定生產者例項名稱
        producer.setInstanceName("producer");
        //啟動生產者
        producer.start();

            //傳送訊息
            for (int i=1;i<=1;i++){
                //模擬網路延遲，每秒傳送一次MQ
                Thread.sleep(1000);
                //建立訊息，topic主題名稱  tags臨時值代表小分類， body代表訊息體
                Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
                //訊息的唯一標識
                message.setKeys("訂單訊息："+i);
                //傳送訊息
                SendResult sendResult=producer.send(message);
                System.out.println("資訊冪等問題來了："+sendResult.toString());
            }
        producer.shutdown();
    }
}

4、消費者

package com.zn.idempotent;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.LogManager;
import java.util.logging.Logger;

/**
 * 訊息冪等消費者
 */
public class IdempotentConsumer {

    static private Map<String, Object> logMap = new HashMap<>();

    public static void main(String[] args) throws MQClientException {
        //建立消費者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //設定NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //設定例項名稱
        consumer.setInstanceName("consumer");
        //訂閱topic
        consumer.subscribe("itmayiedu-topic03","TagA");

        //監聽訊息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                String key=null;
                String msgId=null;

                    for (MessageExt messageExt:list){
                        key=messageExt.getKeys();
                        //判讀redis中有沒有當前訊息key
                        if (logMap.containsKey(key)) {
                            // 無需繼續重試。
                            System.out.println("key:"+key+",已經消費，無需重試...");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        //RocketMQ由於是叢集環境，所以產生的訊息ID可能會重複
                        msgId = messageExt.getMsgId();
                        System.out.println("key:" + key + ",msgid:" + msgId + "---" + new String(messageExt.getBody()));
                        //將當前key儲存在redis中
                        logMap.put(messageExt.getKeys(),messageExt);
                    }
                try {
                    int i=5/0;
                }catch (Exception e){
                    e.printStackTrace();
                    //人工補償
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
        //啟動消費者
        consumer.start();
        System.out.println("Consumer Started!");
    }
}

5、控制檯效果

　　

&n

相關推薦

RocketMQ重試機制和訊息冪等

一、重試機制 　　由於MQ經常處於複雜的分散式系統中，考慮網路波動，服務宕機，程式異常因素，很有可能出現訊息傳送或者消費失敗的問題。因此，訊息的重試就是所有MQ中介軟體必須考慮到的一個關鍵點。如果沒有訊息重試，就可能產生訊息丟失的問題，可能對系統產生很大的影響。所以，秉承寧可多發訊息，也不可丟失訊息的原則，大

C# Command命令(行為型模式)+佇列 實現事務,帶非同步命令重試機制和生命週期

一、簡介 耦合是軟體不能抵禦變變化的根本性原因,不僅實體物件與實體物件之間有耦合關係(如建立性設計模式存在的原因),物件和行為之間也存在耦合關係.   二、實戰 1、常規開發中,我們經常會在控制器中或者Main方法中呼叫多個物件,進行批量的操作(完成一次事務性的操作),像下面這樣:

RocketMQ(5)---RocketMQ重試機制

RocketMQ重試機制 訊息重試分為兩種：Producer傳送訊息的重試 和 Consumer訊息消費的重試。 一、Producer端重試 Producer端重試是指: Producer往MQ上發訊息沒有傳送成功，比如網路原因導致生產者傳送訊息到MQ失敗。 看一下程式碼： @Slf4j public cla

RocketMQ-訊息重試，訊息冪等去重，訊息模式

訊息重試 Rocketmq提供了訊息重試機制，這是一些其他訊息佇列沒有的功能。我們可以依靠這個優秀的機制，而不用在開發中增加更多的業務程式碼去實現 Consumer 消費訊息失敗後，要提供一種重試機制，令訊息再消費一次。Consumer 消費訊息失敗通常可以認為有以下幾種情況  

SpringCloud | FeignClient和Ribbon重試機制區別與聯系

feign per spec 笛卡爾 making log tag tom str 在spring cloud體系項目中，引入的重試機制保證了高可用的同時，也會帶來一些其它的問題，如冪等操作或一些沒必要的重試。 今天就來分別分析一下 FeignClient 和

優雅地處理載入中(loading)，重試(retry)和無資料(empty)等

LoadSir是一個高效易用，低碳環保，擴充套件性良好的載入反饋頁管理框架，在載入網路或其他資料時候，根據需求切換狀態頁面，可新增自定義狀態頁面，如載入中，載入失敗，無資料，網路超時，佔位圖，登入失效等常用頁面。可配合網路載入框架，結合返回狀態碼，錯誤碼，資料進行狀態頁自動切

RocketMQ——訊息冪等問題

一、緣起MQ訊息必達，架構上有兩個核心設計點：（1）訊息落地（2）訊息超時、重傳、確認再次回顧訊息匯流排核心架構，它由傳送端、服務端、固化儲存、接收端四大部分組成。為保證訊息的可達性，超時、重傳、確認機制可能導致訊息匯流排、或者業務方收到重複的訊息，從而對業務產生影響。舉個栗

SpringCloud | FeignClient和Ribbon重試機制區別與聯絡

在spring cloud體系專案中，引入的重試機制保證了高可用的同時，也會帶來一些其它的問題，如冪等操作或一些沒必要的重試。 今天就來分別分析一下 FeignClient 和 Ribbon 重試機制的實現原理和區別，主要分為三點： >1）FeignCl

ActiveMQ的訊息重試機制

處理失敗時的訊息重發機制 1. 處理失敗 指的是MessageListener的onMessage方法裡丟擲RuntimeException。 2. Message頭裡有兩個相關欄位：Redelivered預設為false，redeliveryCounter預設為0。 3.

springboot系列——重試機制原理和應用，還有比這個講的更好的嗎(附完整原始碼)

1. 理解重試機制 2. 總結重試機制使用場景 3. spring-retry重試元件 4. 手寫一個基於註解的重試元件 5. 重試機制下會出現的問題 6. 模板方法設計模式實現非同步重試機制 如果有，請轉給我！ 1. 理解重試機制 “重試是為了提高成功的可能性“ 反過來理解，任何

jedis超時重試機制註意事項

del number 十進制 包含 str 沒有 時間 機制 await 最近使用redis集群進行incr操作，總是發現計數不準確，後來經過檢查發現redis在執行incr超時會執行重試機制，造成計數不準確，測試代碼： /** * incrf: *

guava的重試機制guava-retrying使用

tco exceptio AI ide .class exc erb BE 一個 1，添加maven依賴 <dependency> <groupId>com.github.rholder</groupId> &l

PHP-RESQUE重試機制

pub 實現 方法 ole color except function cti ges 因為PHP-Resque 的重試需要自己寫，網上又沒啥輪子，而且resque也很久不更新了，所以自己研究下resque的源碼，然後也借鑒了Laravel的隊列重試機制，實現了PHP-Re

由表單重復提交引發的冪等性思考

思路 電商 完成後 cor 實現接口 延遲 策略 rom 是否 最近在本地開發測試的時候，遇到一個表單重復提交的現象。其實原因很簡單，因為網絡延遲的問題，我點擊了兩次提交按鈕，數據庫裏生成了兩條記錄。其實這種現象以前也有遇到過，一般都是提交後把按鈕置灰，無法再次提交，這是很

Volley超時重試機制

基礎用法 Volley為開發者提供了可配置的超時重試機制,我們在使用時只需要為我們的Request設定自定義的RetryPolicy即可. 參考設定程式碼如下: int DEFAULT_TIMEOUT_MS = 10000; int DEFAULT_MAX_RETRIES = 3; Str

Appium失敗截圖及重試機制封裝（二）

analyze ret boolean 做了 ktr assert public false fail 一、失敗截圖封裝 1、主要封裝了失敗之後的文件名、重寫了失敗之後消息、失敗了以後做個截圖，最後置為失敗，並且存放到相對路徑下、截圖操作，未把失敗用例至為Fail，主要代

nginx的重試機制 proxy_next_upstream

現在對外服務的網站，很少只使用一個服務節點，而是部署多臺伺服器，上層通過一定機制保證容錯和負載均衡。 nginx就是常用的一種HTTP和反向代理伺服器，支援容錯和負載均衡。 nginx的重試機制就是容錯的一種。 在nginx的配置檔案中，proxy_next_upstream項定義了什麼情況下

Spring Cloud Gateway重試機制

前言 重試，我相信大家並不陌生。在我們呼叫Http介面的時候，總會因為某種原因呼叫失敗，這個時候我們可以通過重試的方式，來重新請求介面。 生活中這樣的事例很多，比如打電話，對方正在通話中啊，訊號不好啊等等原因，你總會打不通，當你第一次沒打通之後，你會打第二次

多執行緒之失敗自動重試機制

發現一個比較好玩的東西: 如果你在使用多執行緒的使用中異常結束了，你應該如何操作呢？ 例子： 正常情況下： 專案一啟動都可以跑完，如果有一段程式碼出現錯誤呢。 系統丟出了一個異常出來。 有沒有發生過這樣的情況，你寫的工作執行緒莫名其妙的掛了，如果不是被你剛好看到，拿只能抓瞎了，不知道啥原因了，因為異常

一個合格程式設計師的標準--------------冪等和非冪等

概念: 在web中的:(下面描述講解的是web)             冪等:                         對於同一種行為，如果執行不論多少次，最終的結果都是一致相同的，就稱這種行為是冪等的。                         (個