欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

RocketMQ重試機制及消息冪代碼實例解析

 更新時間：2020年02月21日 14:36:16   作者：流氓大隊長  
這篇文章主要介紹了RocketMQ重試機制及消息冪代碼實例解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下

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

一.重試機制

　　1.由于MQ經(jīng)常處于復雜的分布式系統(tǒng)中，考慮網(wǎng)絡(luò)波動，服務(wù)宕機，程序異常因素，很有可能出現(xiàn)消息發(fā)送或者消費失敗的問題。因此，消息的重試就是所有MQ中間件必須考慮到的一個關(guān)鍵點。如果沒有消息重試，就可能產(chǎn)生消息丟失的問題，可能對系統(tǒng)產(chǎn)生很大的影響。所以，秉承寧可多發(fā)消息，也不可丟失消息的原則，大部分MQ都對消息重試提供了很好的支持。

　　2.RocketMQ為了使用者封裝了消息重試的處理流程，無需開發(fā)人員手動處理。RocketMQ支持了生產(chǎn)端和消費端兩類重試機制。

模擬異常

　　Consumer端消息消費兩種狀態(tài)：

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.出現(xiàn)異常

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 {
    //創(chuàng)建消費者
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
    //設(shè)置NameServer地址
    consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
    //設(shè)置消費者實例名稱
    consumer.setInstanceName("consumer");
    //訂閱topic
    consumer.subscribe("wn02","TagA");
    //監(jiān)聽消息
    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.網(wǎng)絡(luò)延遲

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 {
    //創(chuàng)建消費者
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
    //設(shè)置NameServer地址
    consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
    //設(shè)置消費者實例名稱
    consumer.setInstanceName("consumer");
    //訂閱topic
    consumer.subscribe("wn03","TagA");
    //監(jiān)聽消息
    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()));
        }

        //網(wǎng)絡(luò)延遲
        try {
          Thread.sleep(600000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
//消息成功
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    consumer.start();
    System.out.println("Consumer Started...");

  }
}

二、消息冪等

1、在什么情況下會發(fā)生RocketMQ的消息重復消費

　　 ①、當系統(tǒng)的調(diào)用鏈路比較長的時候，比如系統(tǒng)A調(diào)用系統(tǒng)B，系統(tǒng)B再把消息發(fā)送到RocketMQ中，在系統(tǒng)A調(diào)用系統(tǒng)B的時候，如果系統(tǒng)B處理成功，但是遲遲沒有將調(diào)用成功的結(jié)果返回給系統(tǒng)A的時候，系統(tǒng)A就會嘗試重新發(fā)起請求給系統(tǒng)B，造成系統(tǒng)B重復處理，發(fā)起多條消息給RocketMQ造成重復消費；

　　②、在系統(tǒng)B發(fā)送給RocketMQ的時候，也有可能會發(fā)生和上面一樣的問題，消息發(fā)送超時，節(jié)骨系統(tǒng)B重試，導致RocketMQ接收到了重讀消息；

　?、邸擱ocketMQ成功接收到消息，并將消息交給消費者處理，如果消費者消費完成后還沒來得及提交offset給RocketMQ，自己宕機或者重啟了，那么RocketMQ沒有接收到offset，就會認為消費失敗了，會重發(fā)消息給消費者再次消費；

2、如何解決消息的重復消費

　　通過冪等性來保證，只要保證重復消息不對結(jié)果產(chǎn)生影響，就完美地解決這個問題。

在生產(chǎn)者端保證冪等性，一下兩種方式：

　　①、RocketMQ支持消息查詢的功能，只要去RocketMQ查詢一下是否已經(jīng)發(fā)送過該條消息就可以了，不存在則發(fā)送，存在則不發(fā)送；

　?、?、引入Redis，在發(fā)送消息到RocketMQ成功之后，向Redis中插入一條數(shù)據(jù)，如果發(fā)送重試，則先去Redis查詢一個該條消息是否已經(jīng)發(fā)送過了，存在的話就不重復發(fā)送消息了；

　　方法一：RocketMQ消息查詢的性能不是特別好，如果在高并發(fā)的場景下，每條消息在發(fā)送到RocketMQ時都去查詢一下，可能會影響接口的性能；

　　方法二：在一些極端的場景下，Redis也無法保證消息發(fā)送成功之后，就一定能寫入Redis成功，比如寫入消息成功而Redis此時宕機，那么再次查詢Redis判斷消息是否已經(jīng)發(fā)送過，是無法得到正確結(jié)果的；

3、生產(chǎn)者

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;

/**
 * 消息冪等生產(chǎn)者
 */
public class IdempotentProvider {
  public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
    //創(chuàng)建一個生產(chǎn)者
    DefaultMQProducer producer=new DefaultMQProducer("rmq-group");
    //設(shè)置NameServer地址
    producer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
    //設(shè)置生產(chǎn)者實例名稱
    producer.setInstanceName("producer");
    //啟動生產(chǎn)者
    producer.start();

      //發(fā)送消息
      for (int i=1;i<=1;i++){
        //模擬網(wǎng)絡(luò)延遲，每秒發(fā)送一次MQ
        Thread.sleep(1000);
        //創(chuàng)建消息，topic主題名稱 tags臨時值代表小分類， body代表消息體
        Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
        //消息的唯一標識
        message.setKeys("訂單消息："+i);
        //發(fā)送消息
        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 {
    //創(chuàng)建消費者
    DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
    //設(shè)置NameServer地址
    consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
    //設(shè)置實例名稱
    consumer.setInstanceName("consumer");
    //訂閱topic
    consumer.subscribe("itmayiedu-topic03","TagA");

    //監(jiān)聽消息
    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)) {
              // 無需繼續(xù)重試。
              System.out.println("key:"+key+",已經(jīng)消費，無需重試...");
              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            //RocketMQ由于是集群環(huán)境，所以產(chǎn)生的消息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!");
  }
}

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

相關(guān)文章

  • java List.of()與Arrays.asList()方法對比分析

    java List.of()與Arrays.asList()方法對比分析

    這篇文章主要為大家介紹了java List.of()與Arrays.asList()方法對比分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪
    2023-11-11
  • JVM 體系結(jié)構(gòu)詳解

    JVM 體系結(jié)構(gòu)詳解

    本文主要介紹了JVM體系結(jié)構(gòu)的相關(guān)知識。具有很好的參考價值。下面跟著小編一起來看下吧
    2017-03-03
  • 聊聊Redis二進制數(shù)組Bitmap

    聊聊Redis二進制數(shù)組Bitmap

    這篇文章主要介紹了Redis二進制數(shù)組Bitmap，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
    2021-07-07
  • 使用spring oauth2框架獲取當前登錄用戶信息的實現(xiàn)代碼

    使用spring oauth2框架獲取當前登錄用戶信息的實現(xiàn)代碼

    這篇文章主要介紹了使用spring oauth2框架獲取當前登錄用戶信息的實現(xiàn)代碼,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
    2020-07-07
  • spring的@Transactional注解用法解讀

    spring的@Transactional注解用法解讀

    這篇文章主要介紹了spring的@Transactional注解用法解讀，具有一定借鑒價值,需要的朋友可以參考下
    2018-01-01
  • List集合按某個屬性或者字段進行分組的操作

    List集合按某個屬性或者字段進行分組的操作

    這篇文章主要介紹了List集合按某個屬性或者字段進行分組的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
    2021-06-06
  • 詳解如何使用SpringBoot封裝Excel生成器

    詳解如何使用SpringBoot封裝Excel生成器

    在軟件開發(fā)過程中，經(jīng)常需要生成Excel文件來導出數(shù)據(jù)或者生成報表，為了簡化開發(fā)流程和提高代碼的可維護性，我們可以使用Spring Boot封裝Excel生成器，本文將介紹如何使用Spring Boot封裝Excel生成器，并提供一些示例代碼來說明其用法和功能
    2023-06-06
  • java常見面試題及答案匯總

    java常見面試題及答案匯總

    本文提供Java面試題集錦,涵蓋封裝、繼承、多態(tài)等核心概念,旨在幫助求職者全面復習,提升面試表現(xiàn),從基礎(chǔ)知識到實際應(yīng)用,內(nèi)容豐富,適合各類求職者,需要的朋友可以參考下
    2024-09-09
  • Java使用HashMap映射實現(xiàn)消費抽獎功能

    Java使用HashMap映射實現(xiàn)消費抽獎功能

    這篇文章主要為大家詳細介紹了Java使用HashMap映射實現(xiàn)消費抽獎功能，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
    2021-09-09
  • 一文秒懂idea的git插件跟翻譯插件

    一文秒懂idea的git插件跟翻譯插件

    idea之類的開發(fā)軟件真的超級多的插件，今天給大家分享idea的git插件跟翻譯插件，感興趣的朋友跟隨小編一起看看吧
    2021-04-04

最新評論