這篇文章主要介紹了RocketMQ重試機制及訊息冪程式碼例項解析,文中通過示例程式碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

一.重試機制

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

　　2.RocketMQ為了使用者封裝了訊息重試的處理流程，無需開發人員手動處理。RocketMQ支援了生產端和消費端兩類重試機制。

模擬異常

　　Consumer端訊息消費兩種狀態：

package com.alibaba.rocketmq.client.consumer.listener;

public enum ConsumeConcurrentlyStatus {
  CONSUME_SUCCESS,RECONSUME_LATER;

  private ConsumeConcurrentlyStatus() {
  }
}

　　一個是成功（CONSUME_SUCCESS），一個是失敗&重試（RECONSUME_LATER）;

　　Consumer為了保證訊息消費成功，只有使用方明確表示消費成功，返回CONSUME_SUCCESS，RocketMQ才會認為訊息消費成功。

　　如果訊息消費失敗，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER,RocketMQ就會認為訊息消費失敗了，需要重新投遞。

　　1.出現異常

package com.wn.consumer;

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;

public class MQConsumer {
  public static void main(String[] args) throws MQClientException {
    //建立消費者
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
    //設定NameServer地址
    consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
    //設定消費者例項名稱
    consumer.setInstanceName("consumer");
    //訂閱topic
    consumer.subscribe("wn02","TagA");
    //監聽訊息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        //獲取訊息
        for (MessageExt msg:list){
          System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
        }

        try {
          int i=1/0;
        }catch (Exception e){
          e.printStackTrace();
          //需要重試
          return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }//訊息成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    consumer.start();
    System.out.println("Consumer Started...");

  }
}

　　2.網路延遲

package com.wn.consumer;

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;

public class MQConsumer {
  public static void main(String[] args) throws MQClientException {
    //建立消費者
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
    //設定NameServer地址
    consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
    //設定消費者例項名稱
    consumer.setInstanceName("consumer");
    //訂閱topic
    consumer.subscribe("wn03",ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        //獲取訊息
        for (MessageExt msg:list){
          System.out.println(msg.getMsgId()+"---"+new String(msg.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!");
  }
}

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。